From hand-wired glue to agents that act

A REST call, end to end

# request: read one order
GET /api/v1/orders/A-1042 HTTP/1.1
Host: api.example.com
Authorization: Bearer <token>
Accept: application/json

# response
HTTP/1.1 200 OK
Content-Type: application/json

{"id": "A-1042", "status": "shipped", "total": 129.90}

The same call in Python:

import requests

r = requests.get(
    "https://api.example.com/api/v1/orders/A-1042",
    headers={"Authorization": "Bearer " + token},
    timeout=10,
)
r.raise_for_status()          # non-2xx becomes an exception
order = r.json()              # {"id": "A-1042", "status": "shipped", ...}

What SOAP looked like for the same job

<soap:Envelope xmlns:soap="http://schemas.xmlsoap.org/soap/envelope/">
  <soap:Body>
    <GetOrder xmlns="urn:orders:v1">
      <OrderId>A-1042</OrderId>
    </GetOrder>
  </soap:Body>
</soap:Envelope>

The WSDL file generated client stubs from this shape. Every change to the contract regenerated and retested the stubs.

Tenant isolation in a query

# every query carries the tenant; a missing filter leaks data across customers
def list_orders(db, tenant_id: str, status: str):
    return db.execute(
        "SELECT id, status, total FROM orders "
        "WHERE tenant_id = %s AND status = %s",
        (tenant_id, status),
    )

A feature flag rollout

if flags.enabled("new_invoice_view", tenant_id=tenant_id):
    return render_invoice_v2(invoice)   # 5% of tenants first, then more
return render_invoice_v1(invoice)

Structured log event

{"event": "workflow_failed", "tenant": "t_204", "step": "invoice.send",
 "error": "smtp_timeout", "release": "2014.06.3", "ms": 30012}

Masked-word prediction, the BERT objective

from transformers import pipeline

fill = pipeline("fill-mask", model="bert-base-uncased")
for p in fill("Invoices are due within thirty [MASK]."):
    print(p["token_str"], round(p["score"], 3))
# days 0.6 ...  (scores are the model's probabilities for each candidate)

Self-attention, the core operation

import numpy as np

def attention(Q, K, V):
    d_k = K.shape[-1]
    scores = Q @ K.T / np.sqrt(d_k)           # how strongly each token attends to each other token
    weights = np.exp(scores - scores.max(-1, keepdims=True))
    weights /= weights.sum(-1, keepdims=True)  # softmax per row
    return weights @ V                       # weighted mix of value vectors

Adapting by prompt instead of retraining

prompt = """Classify the ticket as billing, technical, or other.
Ticket: My card was charged twice.
Label:"""   # one model, no new training run

A complete RAG loop in about 30 lines

import numpy as np

docs = {
    "refund-policy": "Refunds are issued within 14 days of purchase.",
    "shipping": "Orders ship within 2 business days.",
}
# stand-in embedding: swap in a real embedding model in production
def embed(text):
    v = np.zeros(256)
    for ch in text.lower():
        v[ord(ch) % 256] += 1
    return v / np.linalg.norm(v)

index = {doc_id: embed(text) for doc_id, text in docs.items()}

def retrieve(question, k=1):
    q = embed(question)
    ranked = sorted(index, key=lambda d: -float(q @ index[d]))  # cosine similarity
    return ranked[:k]

def build_prompt(question):
    ids = retrieve(question)
    context = "\n".join(f"[{i}] {docs[i]}" for i in ids)
    return f"Answer only from the sources. Cite the id.\n{context}\nQ: {question}"

print(build_prompt("How long do I have to get a refund?"))

Send the printed prompt to any chat model. The model sees only the retrieved passage, and the citation id ties each claim back to the document.

Where it fails

• Bad chunking splits the answer across pieces. Weak retrieval means the right text never reaches the model.

Declaring a tool

tools = [{
    "name": "lookup_order",
    "description": "Return status and total for one order id.",
    "input_schema": {
        "type": "object",
        "properties": {"order_id": {"type": "string"}},
        "required": ["order_id"],
    },
}]

The loop, with the Anthropic Python SDK

import anthropic

client = anthropic.Anthropic()          # reads ANTHROPIC_API_KEY from the environment
MAX_STEPS = 8                            # governance: bounded loop
messages = [{"role": "user", "content": "Has order A-1042 shipped?"}]

def run_tool(name, args):
    if name == "lookup_order":
        return {"status": "shipped", "total": 129.90}
    raise ValueError(f"unknown tool {name}")

for step in range(MAX_STEPS):
    resp = client.messages.create(
        model="<your-model-id>", max_tokens=1024, tools=tools, messages=messages,
    )
    messages.append({"role": "assistant", "content": resp.content})
    if resp.stop_reason != "tool_use":
        break                       # model answered; done
    results = []
    for block in resp.content:
        if block.type == "tool_use":
            out = run_tool(block.name, block.input)    # the system executes the call
            results.append({"type": "tool_result",
                            "tool_use_id": block.id,
                            "content": str(out)})
    messages.append({"role": "user", "content": results})

print(resp.content[-1].text)

The model never runs code itself. It emits a request, the host executes it, and the result goes back as a message.

The wire format: one tool call, as JSON-RPC 2.0

// client to server: discover tools
{"jsonrpc": "2.0", "id": 1, "method": "tools/list"}

// client to server: invoke one
{"jsonrpc": "2.0", "id": 2, "method": "tools/call",
 "params": {"name": "lookup_order", "arguments": {"order_id": "A-1042"}}}

// server to client
{"jsonrpc": "2.0", "id": 2, "result": {
  "content": [{"type": "text", "text": "{\"status\": \"shipped\"}"}],
  "isError": false}}

Writing the server once, with the official Python SDK

from mcp.server.fastmcp import FastMCP

mcp = FastMCP("orders")

# the type hints and docstring become the tool's schema and description
@mcp.tool()
def lookup_order(order_id: str) -> str:
    """Return status and total for one order id."""
    return '{"status": "shipped", "total": 129.90}'

if __name__ == "__main__":
    mcp.run()          # stdio transport by default

Any MCP-capable client can now list and call lookup_order. Adding a second model means one more client, not a second connector.

A skill: a folder with a SKILL.md

refund-review/
├── SKILL.md
└── check_policy.py

---
name: refund-review
description: Review a refund request against the 14-day policy. Use when a customer asks for a refund.
---
1. Load the order with lookup_order.
2. Compute days since purchase. If over 14, deny and cite refund-policy.
3. If 14 or fewer days, draft the refund. Do not send it: stop at the approval gate.

Only the name and description are loaded by default. The steps load when a request matches.

A governance policy, as data

policy = {
    "lookup_order":  "allow",    # read-only
    "draft_refund":  "allow",
    "send_refund":   "ask",      # a person approves each call
    "delete_order":  "deny",
}
MAX_STEPS = 12

def gate(tool_name, args, ask_human):
    decision = policy.get(tool_name, "deny")       # unknown tools are denied
    if decision == "deny":
        return {"error": "not permitted"}
    if decision == "ask" and not ask_human(tool_name, args):
        return {"error": "rejected by reviewer"}
    return run_tool(tool_name, args)

Scheduling and tracing

# cron: run the review every day at 08:00
0 8 * * *  agent run refund-review --trace-dir /var/log/agent

# one trace line per tool call, so every action can be audited
{"run": "r-8831", "step": 3, "tool": "send_refund", "decision": "ask",
 "approved_by": "j.ruiz", "result": "ok", "ts": "2026-11-02T08:00:14Z"}

A minimal eval harness

import json

cases = [json.loads(line) for line in open("cases.jsonl")]
# each line: {"input": "...", "must_call": "lookup_order", "must_contain": "shipped"}

def score(agent, case):
    run = agent(case["input"])                          # returns {"tools": [...], "answer": "..."}
    tool_ok = case["must_call"] in run["tools"]
    text_ok = case["must_contain"].lower() in run["answer"].lower()
    return tool_ok and text_ok

passed = sum(score(agent, c) for c in cases)
rate = passed / len(cases)
print(f"pass rate {rate:.0%} on {len(cases)} cases")

# release gate: block the deploy below the agreed threshold
assert rate >= 0.95, "eval gate failed: do not deploy"

Judging free-text answers

judge_prompt = """Given the source passage and the answer, reply PASS only if every claim
in the answer is supported by the passage. Otherwise reply FAIL and name the claim.

Passage: {passage}
Answer: {answer}"""   # run with a separate model, then spot-check the judge by hand

Tracing a failure

{"run": "r-9012", "steps": [
  {"tool": "lookup_order", "args": {"order_id": "A-1042"}, "ok": true},
  {"tool": "draft_refund", "args": {"order_id": "A-1042", "amount": 129.90}, "ok": false,
   "error": "amount exceeds refundable total"}
], "final": "I could not draft the refund."}

The trace shows which step failed and why. That line goes straight into the case set as a new regression test.