RS
Ringside
Back to cookbook

AI-driven CRM automation

An inbound sales email lands and, seconds later, the deal has moved to the right pipeline stage and a reply is sitting in your outbox waiting for a human to hit send. The activity is already logged. Nobody read the inbox.

The assistant below has four function tools wired to your own database. lookup_contact, update_deal_stage, log_crm_activity and draft_follow_up. Ringside runs the agentic loop and holds the thread state; your server runs the tools and hands the outputs back.

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

What you build: the CRM database (contacts, deals, activities), the tool implementations, the email ingest and the outbox sender.

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. One assistant serves every inbound email.

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

Ringside owns the LLM execution and the 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 on the next iteration without you rebuilding the message array.

Threading each email into its own Ringside thread means the full conversation history, tool inputs and outputs included, is persisted server-side. Retrieve it later for audit or support.

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=ko_0d7f2a91c4e35b86af10d2c7e94b6f3a5d81c02e7b4936af18d5c60e2a7f9b34
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