Capstone part 2: build crescent-crm in Python

Setup

uv init crescent-crm && cd crescent-crm
uv add "mcp[cli]" pydantic
# optional for tests: uv add --dev pytest anyio

This build uses SQLite so you can run it anywhere; swap in Postgres for production by changing the data-access layer. Create a small fictional seed dataset (companies in PK, AE, SA and UK, a few contacts with and without marketing consent, deals in various stages, two months of campaign stats). Never use real customer data for the capstone.

server.py

import os, sqlite3
from contextlib import contextmanager
from datetime import date, datetime, timedelta, timezone
from typing import Annotated, Literal

from pydantic import BaseModel, Field
from mcp.server import MCPServer
from mcp.server.mcpserver.exceptions import ToolError
from mcp.types import ToolAnnotations

DB_PATH = os.environ.get("CRM_DB", "crescent.db")
MAX_ROWS = 25
READ = ToolAnnotations(read_only_hint=True, open_world_hint=False)
WRITE_ADDITIVE = ToolAnnotations(read_only_hint=False, destructive_hint=False, idempotent_hint=True,
                                 open_world_hint=False)
Country = Literal["PK", "AE", "SA", "UK", "US"]
Stage = Literal["lead", "qualified", "proposal", "negotiation", "won", "lost"]

mcp = MCPServer(
    "crescent-crm",
    instructions=("Crescent Growth CRM and campaign data. Start with crm_find_companies to get IDs, "
                  "then crm_company_brief. Money is reported in each record's currency; never convert. "
                  "Writes are limited to logging activities and creating email drafts."),
)

@contextmanager
def db():
    conn = sqlite3.connect(DB_PATH)
    conn.row_factory = sqlite3.Row
    try:
        yield conn
    finally:
        conn.close()

# ---------- models ----------
class CompanyHit(BaseModel):
    company_id: str
    name: str
    industry: str
    country: str

class CompanyList(BaseModel):
    companies: list[CompanyHit]
    note: str = ""

class ContactSummary(BaseModel):
    contact_id: str
    full_name: str
    role: str

class Deal(BaseModel):
    deal_id: str
    name: str
    stage: str
    value: float
    currency: str
    expected_close: str | None
    days_since_update: int

class Activity(BaseModel):
    type: str
    note: str
    created_at: str

class CompanyBrief(BaseModel):
    company: CompanyHit
    key_contacts: list[ContactSummary]
    open_deals: list[Deal]
    recent_activities: list[Activity]
    active_campaigns: list[str]

class Performance(BaseModel):
    start: date
    end: date
    currency_breakdown: dict[str, dict[str, float]]
    previous_period: dict[str, dict[str, float]] | None = None
    note: str = ""

# ---------- read tools ----------
@mcp.tool(title="Find companies", annotations=READ)
def crm_find_companies(
    query: Annotated[str, Field(description="Part of the company name, e.g. 'Noor'")],
    country: Annotated[Country | None, Field(description="Optional country code filter")] = None,
) -> CompanyList:
    """Find client companies by name. Returns IDs to use with crm_company_brief and other tools."""
    with db() as c:
        rows = c.execute("SELECT id, name, industry, country FROM companies WHERE name LIKE ? "
                         "AND (? IS NULL OR country = ?) ORDER BY name LIMIT ?",
                         (f"%{query}%", country, country, MAX_ROWS + 1)).fetchall()
    note = f"Showing first {MAX_ROWS}; refine the query." if len(rows) > MAX_ROWS else ""
    return CompanyList(companies=[CompanyHit(company_id=r["id"], name=r["name"], industry=r["industry"],
                                             country=r["country"]) for r in rows[:MAX_ROWS]], note=note)

@mcp.tool(title="Company brief", annotations=READ)
def crm_company_brief(company_id: Annotated[str, Field(description="ID from crm_find_companies")]) -> CompanyBrief:
    """One-call meeting prep: profile, key contacts (names and roles only), open deals,
    last 5 activities and active campaigns."""
    with db() as c:
        co = c.execute("SELECT id, name, industry, country FROM companies WHERE id=?", (company_id,)).fetchone()
        if not co:
            raise ToolError(f"Unknown company_id {company_id!r}. Use crm_find_companies first.")
        contacts = c.execute("SELECT id, full_name, role FROM contacts WHERE company_id=? "
                             "ORDER BY role LIMIT 10", (company_id,)).fetchall()
        deals = c.execute("SELECT id, name, stage, value, currency, expected_close, updated_at FROM deals "
                          "WHERE company_id=? AND stage NOT IN ('won','lost') ORDER BY value DESC LIMIT 10",
                          (company_id,)).fetchall()
        acts = c.execute("SELECT type, note, created_at FROM activities WHERE company_id=? "
                         "ORDER BY created_at DESC LIMIT 5", (company_id,)).fetchall()
        camps = c.execute("SELECT name FROM campaigns WHERE company_id=? AND end_date >= date('now') "
                          "ORDER BY start_date", (company_id,)).fetchall()
    today = datetime.now(timezone.utc).date()
    return CompanyBrief(
        company=CompanyHit(company_id=co["id"], name=co["name"], industry=co["industry"], country=co["country"]),
        key_contacts=[ContactSummary(contact_id=r["id"], full_name=r["full_name"], role=r["role"]) for r in contacts],
        open_deals=[Deal(deal_id=r["id"], name=r["name"], stage=r["stage"], value=r["value"], currency=r["currency"],
                         expected_close=r["expected_close"],
                         days_since_update=(today - date.fromisoformat(r["updated_at"][:10])).days) for r in deals],
        recent_activities=[Activity(type=r["type"], note=r["note"][:300], created_at=r["created_at"]) for r in acts],
        active_campaigns=[r["name"] for r in camps],
    )

@mcp.tool(title="Find deals", annotations=READ)
def crm_find_deals(
    stage: Stage | None = None,
    min_value: Annotated[float | None, Field(ge=0, description="Minimum deal value in the deal's currency")] = None,
    stale_days: Annotated[int | None, Field(ge=1, le=365, description="Only deals not updated for N days")] = None,
) -> list[Deal]:
    """Find open deals by stage, minimum value and staleness. Returns at most 25, largest first."""
    sql = ("SELECT id, name, stage, value, currency, expected_close, updated_at FROM deals "
           "WHERE stage NOT IN ('won','lost')")
    args: list = []
    if stage:
        sql += " AND stage = ?"; args.append(stage)
    if min_value is not None:
        sql += " AND value >= ?"; args.append(min_value)
    if stale_days:
        sql += " AND updated_at <= ?"; args.append((date.today() - timedelta(days=stale_days)).isoformat())
    sql += " ORDER BY value DESC LIMIT ?"; args.append(MAX_ROWS)
    with db() as c:
        rows = c.execute(sql, args).fetchall()
    today = date.today()
    return [Deal(deal_id=r["id"], name=r["name"], stage=r["stage"], value=r["value"], currency=r["currency"],
                 expected_close=r["expected_close"],
                 days_since_update=(today - date.fromisoformat(r["updated_at"][:10])).days) for r in rows]

def _perf(c, start: date, end: date, market, channel, company_id) -> dict[str, dict[str, float]]:
    rows = c.execute(
        "SELECT k.currency, SUM(d.spend) spend, SUM(d.impressions) imp, SUM(d.clicks) clk, SUM(d.conversions) conv "
        "FROM campaign_daily d JOIN campaigns k ON k.id = d.campaign_id "
        "WHERE d.day BETWEEN ? AND ? AND (? IS NULL OR k.market = ?) AND (? IS NULL OR k.channel = ?) "
        "AND (? IS NULL OR k.company_id = ?) GROUP BY k.currency",
        (start.isoformat(), end.isoformat(), market, market, channel, channel, company_id, company_id)).fetchall()
    out = {}
    for r in rows:
        conv = r["conv"] or 0
        out[r["currency"]] = {"spend": round(r["spend"] or 0, 2), "impressions": r["imp"] or 0,
                              "clicks": r["clk"] or 0, "conversions": conv,
                              "cpa": round((r["spend"] or 0) / conv, 2) if conv else None}
    return out

@mcp.tool(title="Campaign performance", annotations=READ)
def crm_campaign_performance(
    start: date, end: date,
    market: Country | None = None,
    channel: Literal["search", "meta", "tiktok", "linkedin", "youtube", "email"] | None = None,
    company_id: str | None = None,
    compare_previous: Annotated[bool, Field(description="Also return the preceding period of equal length")] = False,
) -> Performance:
    """Aggregated campaign metrics for a date range (max 92 days), grouped by currency."""
    if end < start or (end - start).days > 92:
        raise ToolError("Use a valid range of at most 92 days; split longer periods into several calls.")
    with db() as c:
        current = _perf(c, start, end, market, channel, company_id)
        previous = None
        if compare_previous:
            length = end - start + timedelta(days=1)
            previous = _perf(c, start - length, start - timedelta(days=1), market, channel, company_id)
    return Performance(start=start, end=end, currency_breakdown=current, previous_period=previous,
                       note="" if current else "No campaign data for these filters.")

# ---------- write tools ----------
@mcp.tool(title="Log activity", annotations=WRITE_ADDITIVE)
def crm_log_activity(
    company_id: str,
    type: Literal["call", "meeting", "email", "note"],
    note: Annotated[str, Field(max_length=1000)],
    idempotency_key: Annotated[str, Field(min_length=8, description="Unique per intended log entry; reuse on retry")],
    contact_id: str | None = None,
) -> str:
    """Add an activity to a company's timeline. Additive and idempotent. Requires the crm.write scope."""
    created_by = "mcp-user"   # part 3 replaces this with the authenticated user's identity
    with db() as c:
        if not c.execute("SELECT 1 FROM companies WHERE id=?", (company_id,)).fetchone():
            raise ToolError(f"Unknown company_id {company_id!r}.")
        try:
            c.execute("INSERT INTO activities(id, company_id, contact_id, type, note, created_by, created_at, "
                      "idempotency_key) VALUES (lower(hex(randomblob(8))),?,?,?,?,?,datetime('now'),?)",
                      (company_id, contact_id, type, note, created_by, idempotency_key))
            c.commit()
            return "Activity logged."
        except sqlite3.IntegrityError:
            return "Activity already logged for this idempotency_key (no duplicate created)."

@mcp.tool(title="Draft client email", annotations=WRITE_ADDITIVE)
def crm_draft_client_email(contact_id: str, subject: Annotated[str, Field(max_length=120)],
                           body: Annotated[str, Field(max_length=3000)],
                           purpose: Literal["service", "marketing"]) -> str:
    """Create an email DRAFT for a human to review and send from the CRM. Never sends.
    Marketing drafts require the contact's marketing consent."""
    with db() as c:
        ct = c.execute("SELECT consent_marketing FROM contacts WHERE id=?", (contact_id,)).fetchone()
        if not ct:
            raise ToolError("Unknown contact_id; use crm_company_brief to find contacts.")
        if purpose == "marketing" and not ct["consent_marketing"]:
            raise ToolError("This contact has not consented to marketing emails. Draft a service email or stop.")
        c.execute("INSERT INTO email_drafts(contact_id, subject, body, created_at) VALUES (?,?,?,datetime('now'))",
                  (contact_id, subject, body))
        c.commit()
    return "Draft saved for human review in the CRM drafts queue."

# ---------- resources and prompts ----------
@mcp.resource("crm://schema", mime_type="text/markdown")
def schema_doc() -> str:
    """Tables and metric definitions for crescent-crm."""
    return ("CPA = spend / conversions. CTR = clicks / impressions. Money is in the record's currency. "
            "Deal stages: lead, qualified, proposal, negotiation, won, lost.")

PLAYBOOKS = {"meeting-prep": "1) Brief 2) Open deals and blockers 3) Campaign results 4) Three questions to ask",
             "campaign-retro": "1) Results vs previous period 2) What worked 3) What to change 4) Next tests"}

@mcp.resource("crm://playbooks/{name}", mime_type="text/markdown")
def playbook(name: str) -> str:
    """Agency playbooks: meeting-prep, campaign-retro."""
    if name not in PLAYBOOKS:
        raise ToolError(f"Unknown playbook; choose from {sorted(PLAYBOOKS)}")
    return PLAYBOOKS[name]

@mcp.prompt(title="Account review")
def account_review(company_id: str) -> str:
    """Prepare for a client meeting."""
    return (f"Follow the crm://playbooks/meeting-prep playbook for company {company_id}. "
            "Use crm_company_brief and crm_campaign_performance (last 30 days, compare_previous=true).")

@mcp.prompt(title="Campaign retro")
def campaign_retro(company_id: str, start: str, end: str) -> str:
    """Campaign retrospective for a period."""
    return (f"Run the campaign-retro playbook for {company_id} from {start} to {end} using "
            "crm_campaign_performance with compare_previous=true. Present a table and three recommendations.")

if __name__ == "__main__":
    mcp.run()

(Create an email_drafts(id INTEGER PRIMARY KEY, contact_id, subject, body, created_at) table in your seed script alongside the others.)

Try it

1. Seed the database, then run uv run mcp dev server.py.
2. In the Inspector: call crm_find_companies("Noor"), then crm_company_brief with the returned ID; request a 120-day performance range and confirm the tool error; call crm_log_activity twice with the same key and confirm no duplicate.
3. Connect to Claude Code: claude mcp add crescent-crm -- uv run --with "mcp[cli]" mcp run /absolute/path/server.py and ask: "Brief me on Noor Trading and tell me which of their deals are stale."

Code review notes

• Parameterized SQL everywhere (no string formatting of user input into queries).
• Literals become enums in the schema; Pydantic models become output schemas.
• Notes in the brief are truncated to 300 characters, and contact details never leave the server.
• created_by is a placeholder until part 3 wires in the authenticated identity.