AI SDR

pip install maximem-synap maximem-synap-openai-agents openai-agents celery redis

uv add maximem-synap maximem-synap-openai-agents openai-agents celery redis
# pip-compatible (existing venv): uv pip install maximem-synap maximem-synap-openai-agents openai-agents celery redis

npm install @maximem/synap-js-sdk @maximem/synap-vercel-adk ai @ai-sdk/openai bullmq ioredis zod

from agents import function_tool

@function_tool
async def enrich_prospect(prospect_id: str) -> dict:
    """Fetch role, company, seniority, recent activity from your enrichment provider."""
    return await enrichment.lookup(prospect_id)

@function_tool
async def recent_signals(company_id: str) -> list[dict]:
    """Recent news, hires, funding, product launches at a company. Last 90 days."""
    return await signals.recent(company_id, days=90)

@function_tool
async def get_sequence_state(prospect_id: str) -> dict:
    """Where this prospect is in the sequence: current step, touches sent, replies."""
    return await sequences.state(prospect_id)

const researchTools = {
  enrich_prospect: tool({
    description: "Fetch role, company, seniority, recent activity.",
    parameters: z.object({ prospectId: z.string() }),
    execute: async ({ prospectId }) => enrichment.lookup(prospectId),
  }),
  recent_signals: tool({
    description: "Recent news, hires, funding, product launches at a company.",
    parameters: z.object({ companyId: z.string() }),
    execute: async ({ companyId }) => signals.recent(companyId, 90),
  }),
  get_sequence_state: tool({
    description: "Where this prospect is in the sequence.",
    parameters: z.object({ prospectId: z.string() }),
    execute: async ({ prospectId }) => sequences.state(prospectId),
  }),
};

@function_tool
async def draft_email(prospect_id: str, kind: str, context: dict) -> dict:
    """Return a draft. kind: 'first_touch' | 'follow_up' | 'objection_response' | 'book_meeting'."""
    return {"subject": "...", "body": "...", "draft_id": "..."}

@function_tool
async def send_email(draft_id: str) -> dict:
    """Send the previously-drafted email via your transactional provider."""
    return await email.send_draft(draft_id)

@function_tool
async def book_meeting(prospect_id: str, preferred_times: list[str]) -> dict:
    """Generate a Cal.com link or hold proposed slots on the rep's calendar."""
    return await calendar.book(prospect_id, preferred_times)

@function_tool
async def log_interaction(prospect_id: str, kind: str, details: dict) -> dict:
    """Log a touch into your CRM."""
    return await crm.log(prospect_id, kind, details)

const actionTools = {
  draft_email: tool({
    description: "Draft an email. kind: 'first_touch'|'follow_up'|'objection_response'|'book_meeting'.",
    parameters: z.object({
      prospectId: z.string(), kind: z.string(), context: z.record(z.any()),
    }),
    execute: async ({ prospectId, kind, context }) =>
      drafter.compose(prospectId, kind, context),
  }),
  send_email: tool({
    description: "Send a drafted email.",
    parameters: z.object({ draftId: z.string() }),
    execute: async ({ draftId }) => emailProvider.sendDraft(draftId),
  }),
  book_meeting: tool({
    description: "Generate a booking link or hold slots.",
    parameters: z.object({
      prospectId: z.string(), preferredTimes: z.array(z.string()),
    }),
    execute: async ({ prospectId, preferredTimes }) => calendar.book(prospectId, preferredTimes),
  }),
  log_interaction: tool({
    description: "Log a touch into the CRM.",
    parameters: z.object({
      prospectId: z.string(), kind: z.string(), details: z.record(z.any()),
    }),
    execute: async ({ prospectId, kind, details }) => crm.log(prospectId, kind, details),
  }),
};

SYSTEM = """You are an AI SDR. The user is a prospect or your own internal sequence orchestrator.

When asked to draft a touch:
- Always pull sequence_state, enrich_prospect, and recent_signals first.
- Use prior touches and replies from memory; never re-tread points the prospect ignored or rejected.
- Personalize on real facts. Generic "I see your company is growing" lines are forbidden.
- Match the prospect's prior tone if you have one. Otherwise, be direct, no buzzwords, one ask per email.

When a reply arrives, classify it: interested | objection | out_of_office | unsubscribe | not_now.
- interested → book a meeting via book_meeting.
- objection → store the objection in memory and draft a response addressing it.
- out_of_office → snooze the sequence.
- unsubscribe → kill the sequence immediately, log in CRM.
- not_now → snooze to the date they suggest, or 30 days default.

Always log_interaction with what you did and why."""

async def run_sdr(prospect_id: str, instruction: str) -> str:
    synap_search = create_search_tool(sdk=sdk, user_id=prospect_id, customer_id=CUSTOMER_ID)
    synap_store  = create_store_tool(sdk=sdk, user_id=prospect_id, customer_id=CUSTOMER_ID)

    agent = Agent(
        name="ai_sdr",
        instructions=SYSTEM,
        tools=[
            FunctionTool(synap_search, name_override="synap_search"),
            FunctionTool(synap_store,  name_override="synap_store"),
            enrich_prospect, recent_signals, get_sequence_state,
            draft_email, send_email, book_meeting, log_interaction,
        ],
    )
    result = await Runner.run(agent, input=instruction)
    return result.final_output

const SYSTEM = `You are an AI SDR.

When drafting:
- Always pull sequence_state, enrich_prospect, recent_signals.
- Use prior touches and replies from memory. Never re-tread rejected points.
- Personalize on real facts. No generic openers.
- One ask per email.

When a reply comes in, classify: interested | objection | out_of_office | unsubscribe | not_now.
- interested → book_meeting.
- objection → store memory + draft response.
- out_of_office → snooze.
- unsubscribe → kill sequence + log.
- not_now → snooze 30 days or to suggested date.

Always log_interaction.`;

export async function runSdr(prospectId: string, instruction: string): Promise<string> {
  const model = synap.wrap(openai("gpt-4o"), {
    userId: prospectId,
    customerId: CUSTOMER_ID,
    conversationId: prospectId,
  });

  const { text } = await generateText({
    model,
    system: SYSTEM,
    prompt: instruction,
    tools: { ...researchTools, ...actionTools },
  });
  return text;
}

@celery.task
async def run_due_touches():
    due = await sequences.due_now()  # your DB query
    for prospect_id in due:
        await run_sdr(prospect_id, "It's time for the next touch. Decide and execute.")

// Cron job or BullMQ scheduled worker
async function runDueTouches() {
  const due = await sequences.dueNow();
  for (const prospectId of due) {
    await runSdr(prospectId, "It's time for the next touch. Decide and execute.");
  }
}

async def handle_email_reply(prospect_id: str, body: str, from_email: str):
    await sdk.memories.create(
        document=f"Prospect reply (from {from_email}):\n{body}",
        document_type="prospect-reply",
        user_id=prospect_id,
        customer_id=CUSTOMER_ID,
        metadata={"channel": "email", "direction": "inbound"},
    )
    await run_sdr(prospect_id, "The prospect replied. Read their message and decide.")

export async function handleEmailReply(
  prospectId: string,
  body: string,
  fromEmail: string,
) {
  await synap.sdk.memories.create({
    document: `Prospect reply (from ${fromEmail}):\n${body}`,
    documentType: "prospect-reply",
    userId: prospectId,
    customerId: CUSTOMER_ID,
    metadata: { channel: "email", direction: "inbound" },
  });
  await runSdr(prospectId, "The prospect replied. Read their message and decide.");
}