RS
Ringside
Back to cookbook

AI-driven CRM automation

What you'll build

An inbound-email triage pipeline. When a customer email lands, a webhook fires your handler. Your handler creates a Ringside thread, attaches the email as the first user message, and starts a run against a CRM assistant. The assistant has four function tools wired to your own database: lookup_contact, update_deal_stage, log_crm_activity, and draft_follow_up. Ringside drives the agentic loop; your server executes the tools and submits the outputs back. The run completes with a drafted reply sitting in your outbox and the deal record updated, no human needed for routine handoffs.

What Ringside handles: LLM execution, thread persistence, tool-loop orchestration, per-customer billing, and webhook events when runs complete or fail.

What you build: the CRM database (contacts, deals, activities), the tool implementations, the email ingest, and the outbox sender. Ringside is the AI layer; your app is the CRM layer.

What you need

  • An FC API key with scopes api:chat and api:webhooks
  • pip install openai
  • A Postgres database (or any store) with contacts, deals, and activities tables
  • An HTTPS endpoint for Ringside webhooks (use ngrok in dev)

Full code

1. Create the CRM assistant (one-time setup)

python
# setup_assistant.py
import os
from openai import OpenAI

client = OpenAI(api_key=os.environ["FC_API_KEY"], base_url="https://api.fightclub.pro/v1")

assistant = client.beta.assistants.create(
    name="CRM Triage Agent",
    instructions="""You are an AI CRM assistant. When given an inbound email:
1. Look up the sender's contact record.
2. Identify the relevant deal (by company or subject line context).
3. Update the deal stage if the email signals a clear progression (e.g. "we're ready to sign" → Negotiation).
4. Log the email as a CRM activity.
5. Draft a concise, professional follow-up reply appropriate for the deal stage.
Always call lookup_contact first. Never invent contact or deal ids.""",
    model="fc:openai/gpt-4o",
    tools=[
        {
            "type": "function",
            "function": {
                "name": "lookup_contact",
                "description": "Look up a contact by email address. Returns contact id, name, company, and their open deals.",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "email": {"type": "string", "description": "Sender email address"},
                    },
                    "required": ["email"],
                },
            },
        },
        {
            "type": "function",
            "function": {
                "name": "update_deal_stage",
                "description": "Move a deal to a new pipeline stage.",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "deal_id": {"type": "string"},
                        "new_stage": {
                            "type": "string",
                            "enum": ["Lead", "Qualified", "Proposal", "Negotiation", "Closed Won", "Closed Lost"],
                        },
                        "reason": {"type": "string", "description": "One-line reason for the stage change"},
                    },
                    "required": ["deal_id", "new_stage", "reason"],
                },
            },
        },
        {
            "type": "function",
            "function": {
                "name": "log_crm_activity",
                "description": "Append an activity (email, note, call) to a contact's timeline.",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "contact_id": {"type": "string"},
                        "activity_type": {"type": "string", "enum": ["email", "note", "call"]},
                        "summary": {"type": "string", "description": "Brief description of the activity"},
                    },
                    "required": ["contact_id", "activity_type", "summary"],
                },
            },
        },
        {
            "type": "function",
            "function": {
                "name": "draft_follow_up",
                "description": "Store a drafted reply in the outbox. The caller sends it after human review, or auto-sends for low-value leads.",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "contact_id": {"type": "string"},
                        "subject": {"type": "string"},
                        "body": {"type": "string"},
                        "send_immediately": {
                            "type": "boolean",
                            "description": "True only for automated low-priority acknowledgements",
                        },
                    },
                    "required": ["contact_id", "subject", "body", "send_immediately"],
                },
            },
        },
    ],
)

print(f"ASSISTANT_ID={assistant.id}")
# Persist this id in your env — you reuse one assistant for all inbound emails.

2. Inbound email handler

python
# handler.py
import json, os, time
from openai import OpenAI
import db  # your ORM / db module

client = OpenAI(api_key=os.environ["FC_API_KEY"], base_url="https://api.fightclub.pro/v1")
ASSISTANT_ID = os.environ["CRM_ASSISTANT_ID"]


def handle_inbound_email(email: dict) -> None:
    """
    email = {
        "from": "jane@acme.com",
        "subject": "Re: proposal — we're ready to move forward",
        "body": "Hi team, we reviewed the proposal and we're ready to sign...",
    }
    """

    # Create a thread scoped to this customer (tracked for billing).
    thread = client.beta.threads.create(
        metadata={"customer_external_id": f"ext:{email['from']}"},
    )

    # Attach the email as the user turn.
    client.beta.threads.messages.create(
        thread_id=thread.id,
        role="user",
        content=f"From: {email['from']}\nSubject: {email['subject']}\n\n{email['body']}",
    )

    # Start the run.
    run = client.beta.threads.runs.create(
        thread_id=thread.id,
        assistant_id=ASSISTANT_ID,
    )

    # Drive the tool loop.
    while run.status in ("queued", "in_progress", "requires_action"):
        if run.status == "requires_action":
            outputs = []
            for call in run.required_action.submit_tool_outputs.tool_calls:
                result = dispatch_tool(call.function.name, json.loads(call.function.arguments))
                outputs.append({"tool_call_id": call.id, "output": json.dumps(result)})
            run = client.beta.threads.runs.submit_tool_outputs(
                thread_id=thread.id, run_id=run.id, tool_outputs=outputs,
            )
            continue
        time.sleep(0.5)
        run = client.beta.threads.runs.retrieve(thread_id=thread.id, run_id=run.id)

    if run.status != "completed":
        raise RuntimeError(f"Run ended with status={run.status}: {run.last_error}")

    # The final assistant message is a human-readable summary — log it.
    msgs = client.beta.threads.messages.list(thread_id=thread.id, order="desc", limit=1)
    print("Agent summary:", msgs.data[0].content[0].text.value)


def dispatch_tool(name: str, args: dict) -> dict:
    if name == "lookup_contact":
        contact = db.contacts.find_by_email(args["email"])
        if not contact:
            return {"error": "contact_not_found"}
        deals = db.deals.find_open_by_contact(contact["id"])
        return {"contact_id": contact["id"], "name": contact["name"], "company": contact["company"], "open_deals": deals}

    if name == "update_deal_stage":
        db.deals.update_stage(args["deal_id"], args["new_stage"])
        db.activities.log(args["deal_id"], "stage_change", args["reason"])
        return {"ok": True, "deal_id": args["deal_id"], "new_stage": args["new_stage"]}

    if name == "log_crm_activity":
        db.activities.log_contact(args["contact_id"], args["activity_type"], args["summary"])
        return {"ok": True}

    if name == "draft_follow_up":
        db.outbox.insert(args["contact_id"], args["subject"], args["body"])
        if args.get("send_immediately"):
            db.outbox.send_now(args["contact_id"])
        return {"ok": True, "queued_for_review": not args.get("send_immediately")}

    return {"error": f"unknown tool: {name}"}

3. Register the run.completed webhook (optional)

Register a webhook so Ringside notifies you when a run finishes asynchronously, useful if you want to decouple the HTTP response from the agent loop:

bash
curl -X POST https://api.fightclub.pro/v1/webhooks \
  -H "Authorization: Bearer $FC_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "url": "https://your-app.com/webhooks/ringside",
    "events": ["run.completed", "run.failed"]
  }'

Verify the HMAC signature on delivery:

python
import hashlib, hmac, time

def verify_ringside_webhook(body: bytes, signature_header: str, secret: str) -> bool:
    ts, v1 = (p.split("=", 1)[1] for p in signature_header.split(","))
    if abs(time.time() - int(ts)) > 300:
        return False   # replay window
    expected = hmac.new(secret.encode(), f"{ts}.".encode() + body, hashlib.sha256).hexdigest()
    return hmac.compare_digest(expected, v1)

Walkthrough

The key insight is the split of responsibility: Ringside owns the LLM execution and tool-loop state machine; your server owns the data. requires_action is Ringside telling you "the model wants to call a function, here are the arguments, give me the outputs". You call your database, return a JSON blob, and Ringside feeds it back into the model's context automatically.

Threading each email into its own Ringside thread means the full conversation history (including all tool inputs/outputs) is persisted server-side. You can retrieve it later for audit, customer support, or fine-tuning input.

The metadata.customer_external_id on thread-create routes billing to the correct Ringside Customer, so you can see cost-per-contact in the dashboard without any extra instrumentation.

Run the Assistants shim against the same assistant id for every email: one assistant definition, unlimited threads. The 10-iteration tool-call cap means a runaway model can't rack up unbounded API costs on a single email.

Run it

bash
export FC_API_KEY=sk_live_xxx
export CRM_ASSISTANT_ID=asst_xxx   # printed by setup_assistant.py

# One-time setup:
python setup_assistant.py

# Simulate an inbound email:
python -c "
import handler
handler.handle_inbound_email({
    'from': 'jane@acme.com',
    'subject': 'Re: proposal — ready to sign',
    'body': 'Hi, we reviewed the proposal and are ready to move forward. Can we schedule a call?',
})
"

You should see a deal stage update, an activity logged, a drafted reply in your outbox, and an agent summary printed to stdout.

What's next