This module

In all cases, the errors in the output can be fed back to the model for correction

SYSTEM = """You are a BigQuery SQL writer.
Return ONLY the SQL — nothing else.
- No markdown, no ```sql code fences
- No commentary, no explanation
- Lowercase keywords (select, from, where)
- Always include a LIMIT clause
- Quote reserved identifiers with backticks"""

@app.route("/api/sql", methods=["POST"])
def generate_sql():
    schema_dump = request.json["schema_dump"]
    question    = request.json["question"]

    resp = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=800,
        system=SYSTEM + f"\n\nSCHEMA:\n{schema_dump}",
        messages=[{"role": "user", "content": question}],
    )

    raw = resp.content[0].text.strip()
    # Defensive strip — model occasionally ignores the no-fence rule
    if raw.startswith("```"):
        raw = raw.split("\n", 1)[1].rsplit("```", 1)[0].strip()
    return {"sql": raw}

const schema_dump = `
TABLE \`acme.sales.orders\` (
  order_id     STRING,
  customer_id  STRING,
  order_date   DATE,
  total_usd    NUMERIC,
  status       STRING  -- pending | shipped | delivered | cancelled
)`

const question = "How much did we sell in the last 7 days, by status?"

const res = await fetch('/api/sql', {
  method: 'POST',
  headers: { 'Content-Type': 'application/json' },
  body: JSON.stringify({ schema_dump, question }),
})
const { sql } = await res.json()
console.log(sql)

select status, sum(total_usd) as total_sales
from `acme.sales.orders`
where order_date >= date_sub(current_date(), interval 7 day)
group by status
limit 100

One LLM call. One string out. The "structure" is whatever you trust the prompt + strip to produce. Cheap, fast, easy to reason about — but no machine-checkable guarantees. A typo in a column name, a missing LIMIT, or a stray ``` fence — your downstream code has to handle it.

Don't forget to check for compliance and pass descriptive errors back into the model to try again (max retries before failure)

from pydantic import BaseModel, EmailStr, \
     ValidationError, Field

MAX_ATTEMPTS = 3

class Person(BaseModel):
    name: str = Field(min_length=1)
    email: EmailStr
    company: str | None = None

@app.route("/api/extract-person", methods=["POST"])
def extract_person():
    text = request.json["text"]
    messages = [{"role": "user", "content": text}]

    for attempt in range(MAX_ATTEMPTS):
        resp = client.messages.create(
            model="claude-sonnet-4-6",
            max_tokens=512,
            system="Extract the person's name, "
                   "email, and company.",
            messages=messages,
            tools=[EXTRACT_PERSON],
            tool_choice={"type": "tool",
                         "name": "record_person"},
        )
        block = next(b for b in resp.content
                     if b.type == "tool_use")

        try:
            person = Person(**block.input)
            return {"person": person.model_dump()}  # 200
        except ValidationError as e:
            # Show the model exactly what went wrong
            messages.append({"role": "assistant",
                             "content": resp.content})
            messages.append({"role": "user", "content": [{
                "type": "tool_result",
                "tool_use_id": block.id,
                "is_error": True,
                "content": (
                    f"Validation failed:\n{e}\n\n"
                    "Re-call record_person with "
                    "corrected fields."
                ),
            }]})

    # All attempts exhausted — structured 422
    return jsonify({
        "error": "extraction_failed",
        "message": (
            f"Could not extract a valid person "
            f"after {MAX_ATTEMPTS} attempts."
        ),
        "last_validation_error": str(e),
    }), 422

The endpoint's contract: 200 with a validated person, or 422 with a structured error. Callers never see a malformed person. The retry budget caps cost at MAX_ATTEMPTS LLM calls per request.

{
  "person": {
    "name": "Sam",
    "email": "sam@acme.com",
    "company": "Acme"
  }
}

{
  "error": "extraction_failed",
  "message": "Could not extract a valid person
              after 3 attempts.",
  "last_validation_error": "1 validation error
      for Person\nemail\n  value is not a valid
      email address [type=value_error]"
}

In practice the second attempt almost always succeeds — the is_error: true tool_result is a clear "here's what went wrong, try again" signal the model treats as a correction, not a generic failure.

def agent(messages, tools):
    while True:
        resp = client.messages.create(
            model="claude-sonnet-4-6",
            max_tokens=2048,
            system="...",
            messages=messages,
            tools=tools, 
        )
        messages.append({"role": "assistant", "content": resp.content})

        if resp.stop_reason == "end_turn":
            return resp

        # Run every tool the model asked for, feed the results back
        results = []
        for block in resp.content:
            if block.type == "tool_use":
                out = run_tool(block.name, block.input)
                results.append({
                    "type": "tool_result",
                    "tool_use_id": block.id,
                    "content": out,
                })
        messages.append({"role": "user", "content": results})

Each iteration: model proposes tool calls → you run them → results go back as a tool_result message → loop. Terminate when stop_reason == "end_turn". This same ~20-line loop is the engine inside Claude Code, Cursor, every agent framework. Model decides when it is done (can meet the users request or hits a dead end)

turn 1  →  model:  tool_use grep(pattern="/api/chat", path=".")
        ←  tool:   backend/app.py:18: @app.route("/api/chat", methods=["POST"])
                   frontend/src/App.jsx:14:    const res = await fetch('/api/chat', {

turn 2  →  model:  tool_use read_file(path="backend/app.py", start=1, end=40)
        ←  tool:   from anthropic import Anthropic
                   ...
                   @app.route("/api/chat", methods=["POST"])
                   def chat():
                       user_message = request.json["message"]
                       resp = client.messages.create(
                           model="claude-sonnet-4-6",
                           ...
                       )
                       return jsonify({"reply": resp.content[0].text})

turn 3  →  model:  end_turn

  ASSISTANT FINAL ANSWER
  ─────────────────────────────────────────────────────────────────
  The chat endpoint is defined in `backend/app.py:18`. It forwards the
  user's message to Anthropic's API via `client.messages.create(...)` on
  line 22, using model `claude-sonnet-4-6`. The reply is returned as
  JSON under the `reply` key.

Three turns, two tool calls, one citation-grounded answer. The model chose to grep first to locate the route, then read_file for context. Swap the tools for get_order / refund / send_email and you have a support agent. The shape is the same.

Trust but verify. Every production prompt has guardrails on both sides — what you send and what comes back.

# 1. Hand-write the schema
GITHUB_LIST_ISSUES = {
    "name": "github_list_issues",
    "description": "List issues for a repo.",
    "input_schema": {
        "type": "object",
        "properties": {
            "owner": {"type": "string"},
            "repo":  {"type": "string"},
            "state": {"type": "string", "enum": ["open", "closed", "all"]},
        },
        "required": ["owner", "repo"],
    },
}

# 2. Hand-write the glue — auth, request, response shaping
def github_list_issues(owner, repo, state="open"):
    r = requests.get(
        f"https://api.github.com/repos/{owner}/{repo}/issues",
        headers={"Authorization": f"Bearer {os.environ['GH_TOKEN']}"},
        params={"state": state},
    )
    r.raise_for_status()
    return [{"number": i["number"], "title": i["title"]} for i in r.json()]

This works. It doesn't scale. N apps × M APIs = N × M hand-rolled implementations.

The hand-rolled approach grows as a product (N × M). MCP collapses it into a sum (N + M) — because the work moves to the side that owns the API.

Think of it as HTTP for LLM tools. The protocol doesn't tell you which servers exist — but once you point a client at one, everything else is standardized.

# pip install mcp
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("demo-server")

@mcp.tool()
def add(a: int, b: int) -> int:
    """Add two numbers together."""
    return a + b

@mcp.tool()
def greet(name: str) -> str:
    """Return a friendly greeting."""
    return f"Hello, {name}!"

@mcp.resource("config://app")
def get_config() -> str:
    """Expose some static config as a readable resource."""
    return "app_name=demo\nversion=0.1"

if __name__ == "__main__":
    mcp.run()    # defaults to stdio transport

The @mcp.tool() decorator generates the schema from type hints and the description from the docstring. @mcp.resource(uri) exposes read-only data. Compare to the hand-rolled GitHub example — that was one tool. This is two tools plus a resource, with no schema written by hand.

import asyncio
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

async def main():
    # Tell the client how to launch the server process
    params = StdioServerParameters(command="python", args=["server.py"])

    async with stdio_client(params) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()

            # DISCOVERY — the schemas come from the server, not from your code
            tools = await session.list_tools()
            print("Tools:", [t.name for t in tools.tools])
            #   → Tools: ['add', 'greet']

            resources = await session.list_resources()
            print("Resources:", [r.uri for r in resources.resources])
            #   → Resources: ['config://app']

            # INVOCATION
            result = await session.call_tool("add", {"a": 2, "b": 3})
            print("add(2,3) =", result.content[0].text)        # → 5

            cfg = await session.read_resource("config://app")
            print(cfg.contents[0].text)                        # → app_name=demo

asyncio.run(main())

The client never hardcodes the tool list. If the server author adds a new tool tomorrow, this same client picks it up on the next list_tools() call. That is the win — the integration is no longer something you maintain.

async with ClientSession(read, write) as session:
    await session.initialize()
    discovered = await session.list_tools()

    # Translate MCP tool schemas → Anthropic tool-use format
    anthropic_tools = [
        {"name": t.name, "description": t.description, "input_schema": t.inputSchema}
        for t in discovered.tools
    ]

    resp = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=1024,
        tools=anthropic_tools,                         # ← discovered, not hardcoded
        messages=[{"role": "user", "content": "What's 17 + 25?"}],
    )

    for block in resp.content:
        if block.type == "tool_use":
            out = await session.call_tool(block.name, block.input)   # ← MCP handles it
            # …feed the result back into the next messages.create call

MCP handles discovery and invocation. The LLM layer handles reasoning about when to call what. Same agent loop you saw earlier — the tool registry is just sourced over MCP instead of hand-coded.

{
  "mcpServers": {
    "demo": { "command": "python", "args": ["/path/to/server.py"] }
  }
}

External directories (Anthropic's catalog, Smithery, community lists) tell you which servers exist — but that's a layer on top of the protocol, not in it. Like REST: HTTP doesn't tell you which APIs exist; you find them through docs.

Alternative welcome. Propose your own structured-extraction use case if it's more interesting to you. Present: your schema, your system prompt, one case that works, one that broke, and what you did about it.