--- title: "Capstone part 2: build crescent-crm in Python" description: "Setup 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…" url: https://optimizeall.com/learn/model-context-protocol-mcp/capstone-build-crm-mcp-server updated: 2026-10-05 --- Model Context Protocol (MCP): Connect AI to Your Tools and Data · Capstone: an MCP server for a marketing CRM · lesson 17 of 18 · 24 min # Capstone part 2: build crescent-crm in Python ## Setup ```bash 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 ```python 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. ## Video lecture: Capstone part 2: build crescent-crm in Python Lecture coming soon · 15 chapters · about 9 minutes. Read the full transcript below. 1. Capstone part 2: build 2. Why build carefully? 3. Setup 4. Foundations 5. Read tools 6. Simple example: the unknown ID 7. More read tools 8. Safe writes 9. Resources and prompts 10. Try it 11. Dates the model gets wrong 12. Deeper: a realistic session 13. Watch me do it: key build pieces 14. Try this now 15. Recap ## Lecture transcript ### Capstone part 2: build Time to build. In this lesson you'll turn the crescent CRM design into a working Python server with the official SDK: five read tools, two safe write tools, two resources and two prompts, running on a fictional SQLite dataset you can swap for Postgres later. ### Why build carefully? Why build it with such care when it's a capstone? Because the habits you practice here are the habits that matter in production: parameterized queries, capped results, helpful errors, idempotent writes and data minimization. Think of practicing scales on a piano. They're not the concert, but every concert depends on them. By the end of this lesson, those habits will be in your fingers, and you'll write your next server this way without thinking. ### Setup Set up with uv: create the project and add the MCP SDK with its command line extra, plus Pydantic. Then create a small fictional seed dataset: companies in Pakistan, the UAE, Saudi Arabia and the UK, contacts with and without marketing consent, deals at different stages, and two months of campaign stats. Never use real customer data for a learning project. The server reads the database path from an environment variable. ### Foundations Look at the foundations first. The server's instructions tell hosts to start with find companies, then the brief, and to never convert currencies. Literal types define countries, deal stages and channels, so they become enums in the schema. Annotations mark read tools as read only, and write tools as additive and idempotent. Pydantic models define every output, so clients receive structured, validated data. ### Read tools Now the read tools. Find companies searches by name with an optional country, returns at most twenty five, and adds a note when there are more. The company brief is the workhorse: profile, up to ten contacts with names and roles only, open deals with days since last update, the last five activities with notes cut to three hundred characters, and active campaigns. Unknown ids raise a tool error that says use find companies first. ### Simple example: the unknown ID A simple example from the code. Look at what happens when the model calls company brief with a made up id, like co underscore acme two. The tool doesn't crash and doesn't return an empty brief. It raises a tool error saying, unknown company id, use find companies first. The model reads that, calls find companies with the name, gets the real id, and tries again. One helpful sentence turns a dead end into a two step recovery. ### More read tools Find deals filters by stage, minimum value and staleness, building a parameterized query step by step. Campaign performance aggregates metrics by currency for a date range of up to ninety two days, and can also return the previous period of equal length for comparison. If the range is too long, it raises a tool error telling the model to split it into several calls. Every query uses placeholders, never string formatting, because tool arguments come from a model that untrusted text can influence. ### Safe writes Two safe write tools. Log activity adds to a company's timeline, and it's idempotent: a unique constraint on the idempotency key means a retry returns, already logged, instead of creating a duplicate. Draft client email saves a draft for a human to review and send from the CRM. It never sends. And for marketing emails, it checks consent first and refuses if the contact hasn't opted in. That rule lives in code, so no clever prompt can talk the model around it. ### Resources and prompts Resources and prompts round it out. The schema resource explains metrics like cost per acquisition and click through rate. The playbooks resource template serves meeting prep and campaign retro guides. The account review prompt follows the meeting prep playbook using the brief and thirty days of performance, and the campaign retro prompt compares periods and asks for three recommendations. ### Try it Now try it. Seed the database and open the Inspector with mcp dev. Find a company, then get its brief. Ask for a hundred and twenty day performance range and confirm you get a helpful error. Log the same activity twice with the same key and confirm there's no duplicate. Then connect it to Claude Code and ask, brief me on Noor Trading and tell me which deals are stale. Watch which tools the model chooses. ### Dates the model gets wrong What if the model asks for a date range in the wrong format, or in words, like last month? The Pydantic date type rejects invalid formats automatically with a clear validation message, and the model usually retries correctly. For relative ranges, you have two options: add a helper tool that converts phrases like last month into exact dates, or state in the description that dates must be absolute, in year month day format. Either way, test it with a few natural phrasings. ### Deeper: a realistic session Let's deepen the build with a realistic test session for Crescent Growth. An account manager asks Claude Code, brief me on Al Noor Trading and tell me which deals are stale. The model calls find companies with Noor, gets the id, calls company brief, and notices a proposal untouched for forty days. It then asks, should I log that you'll follow up on Friday? The manager says yes, and the model calls log activity with a fresh idempotency key. The network hiccups, the host retries, and the second call returns, already logged. The CRM shows exactly one activity, attributed correctly. That small sequence exercises nearly every design decision you made in part one. ### Watch me do it: key build pieces Watch me do it. Let's step through three pieces of the build. First, the database helper: a context manager opens SQLite with row objects and always closes the connection. Second, company brief: it looks up the company and raises a tool error if missing. Then four queries: up to ten contacts with only id, name and role; up to ten open deals, largest first; the last five activities; and active campaigns. It computes days since update, truncates notes to three hundred characters, and returns the brief model. Third, log activity: it checks the company exists, then inserts the activity with the idempotency key. If SQLite raises an integrity error because the key already exists, the tool returns, already logged, instead of failing. Now I run the four Inspector checks: brief works, the hundred and twenty day range errors, the duplicate log is harmless, and a marketing draft to a non consented contact is refused. ### Try this now Try this now. Create a small fictional dataset, build the server, and run the four Inspector checks: find a company and get its brief, request a hundred and twenty day range and read the error, log the same activity twice with the same key, and try a marketing draft to a contact without consent. Then connect it to Claude Code and ask a real question. Note which tools the model calls and in what order. That order tells you whether your descriptions are doing their job. ### Recap To recap: you've built a server with compact, structured read tools, idempotent and consent aware writes, parameterized queries, caps and truncation, plus resources and prompts. One placeholder remains: the created by field, which part three connects to the authenticated user. Your next step: build it with fictional data, run the four Inspector checks, and connect it to Claude Code or VS Code. ## Key takeaways - The capstone server implements five read tools, two safe write tools, two resources and two prompts. - Parameterized SQL, row caps, range limits and truncation keep outputs safe and compact. - Idempotency keys with a unique constraint make activity logging retry-safe. - Consent checks enforce data-protection rules inside the tool, not in the prompt. - Test interactively in the Inspector, then connect a real host. ## Try it Build crescent-crm with a fictional seed dataset, verify the four Inspector checks, and connect it to Claude Code or VS Code. - [Previous: Capstone part 1: design an MCP server for a marketing CRM dataset](https://optimizeall.com/learn/model-context-protocol-mcp/capstone-design-crm-mcp-server) - [Next: Capstone part 3: secure, test and ship crescent-crm](https://optimizeall.com/learn/model-context-protocol-mcp/capstone-secure-test-ship-crm-mcp) - [All lessons of Model Context Protocol (MCP): Connect AI to Your Tools and Data](https://optimizeall.com/learn/model-context-protocol-mcp)