AI agents and MCP

Lesson 20 — Python

Lesson 20 of 20 Advanced ~110 min

Learning objectives

• Define what an agent is, and when one is warranted
• Build an agent loop from first principles
• Orchestrate multi-step work with LangGraph
• Expose tools over the Model Context Protocol
• Apply guardrails, budgets and human-in-the-loop controls
• Judge honestly whether an agent belongs in a regulated workflow

What an agent is

A chain has a fixed sequence of steps. An agent decides its own next step: it observes, chooses a tool, sees the result, and repeats until it judges the task done.

        ┌──────────────────────────────┐
        ▼                              │
   Observe ──▶ Decide ──▶ Act ──▶ Result
                  │
                  └──▶ Done ──▶ Answer

That loop is the whole idea. Everything else — frameworks, protocols, orchestration graphs — is machinery for making it reliable.

ImportantMost tasks do not need an agent

An agent trades determinism for flexibility. In exchange for handling problems you did not anticipate, you get variable cost, variable latency, and an execution path that differs between runs.

Use a chain	Use an agent
Steps are known in advance	The steps depend on what is found
Same shape every time	Genuinely open-ended investigation
Reproducibility matters	Exploration matters more
Regulated output	Analyst assistance

If you can write the steps down, write the steps down. A chain is cheaper, faster, testable and auditable. Reach for an agent when the task genuinely requires choosing what to do next based on intermediate results.

An agent from first principles

Worth building once, because every framework is this loop with extra layers.

from __future__ import annotations

import json
import logging
from dataclasses import dataclass, field
from typing import Any, Callable

logger = logging.getLogger(__name__)


@dataclass
class Tool:
    """A function the agent may call, with a schema the model can read."""
    name: str
    description: str
    parameters: dict
    func: Callable[..., Any]
    requires_approval: bool = False

    def to_schema(self) -> dict:
        return {"type": "function", "name": self.name,
                "description": self.description, "parameters": self.parameters}


@dataclass
class AgentResult:
    answer: str
    steps: list[dict] = field(default_factory=list)
    input_tokens: int = 0
    output_tokens: int = 0
    halted_reason: str | None = None


class Agent:
    """A minimal, bounded tool-using agent."""

    def __init__(self, client, model: str, tools: list[Tool], system: str,
                 max_steps: int = 10, max_tokens_total: int = 100_000):
        self.client = client
        self.model = model
        self.tools = {t.name: t for t in tools}
        self.system = system
        self.max_steps = max_steps
        self.max_tokens_total = max_tokens_total

    def run(self, task: str, approve: Callable[[str, dict], bool] | None = None
            ) -> AgentResult:
        messages: list[dict] = [{"role": "user", "content": task}]
        result = AgentResult(answer="")

        for step in range(1, self.max_steps + 1):
            if result.input_tokens + result.output_tokens > self.max_tokens_total:
                result.halted_reason = "token budget exhausted"
                break

            response = self.client.responses.create(
                model=self.model,
                instructions=self.system,
                input=messages,
                tools=[t.to_schema() for t in self.tools.values()],
                temperature=0,
            )
            result.input_tokens += response.usage.input_tokens
            result.output_tokens += response.usage.output_tokens

            calls = [i for i in response.output if i.type == "function_call"]

            if not calls:                              # the model is finished
                result.answer = response.output_text
                return result

            messages += response.output

            for call in calls:
                record = {"step": step, "tool": call.name,
                          "arguments": call.arguments}

                if call.name not in self.tools:         # allowlist
                    output = {"error": f"Unknown tool '{call.name}'"}
                else:
                    tool = self.tools[call.name]
                    args = json.loads(call.arguments)

                    if tool.requires_approval:
                        if approve is None or not approve(tool.name, args):
                            output = {"error": "Rejected by the operator"}
                            record["approved"] = False
                        else:
                            record["approved"] = True
                            output = self._invoke(tool, args)
                    else:
                        output = self._invoke(tool, args)

                record["result"] = str(output)[:400]
                result.steps.append(record)
                logger.info("Step %d: %s -> %s", step, call.name,
                            str(output)[:120])

                messages.append({
                    "type": "function_call_output",
                    "call_id": call.call_id,
                    "output": json.dumps(output, default=str),
                })

        result.halted_reason = result.halted_reason or "step limit reached"
        result.answer = "The agent did not reach a conclusion within its limits."
        return result

    @staticmethod
    def _invoke(tool: Tool, args: dict) -> Any:
        """Tool errors are returned to the model, not raised."""
        try:
            return tool.func(**args)
        except Exception as e:
            logger.warning("Tool %s failed: %s", tool.name, e)
            return {"error": f"{type(e).__name__}: {e}"}

Four things in that loop matter more than the rest:

A step limit. Without one, a confused agent loops until the budget or your patience runs out. Ten is a reasonable default; if a task genuinely needs thirty, it probably needs decomposing.

A token budget. The step limit bounds iterations, not cost. A single step with a large retrieved context can be expensive.

An allowlist. The model returns a tool name. Looking it up in a dictionary you control means a hallucinated or injected tool name fails safely.

Tool errors returned, not raised. An agent that receives {"error": "no such subject"} can correct itself. One that crashes cannot.

Tools

import pandas as pd

def count_subjects(arm: str, min_age: int = 0) -> dict:
    """Read-only, enumerated, validated."""
    if arm not in {"Placebo", "Drug A", "Drug B", "All"}:
        return {"error": f"Unknown arm '{arm}'"}
    d = adsl if arm == "All" else adsl[adsl["TRT01A"] == arm]
    d = d[d["AGE"] >= min_age]
    return {"arm": arm, "min_age": min_age, "n_subjects": int(len(d))}


def summarise_parameter(paramcd: str, visit: int) -> dict:
    if paramcd not in set(adlb["PARAMCD"]):
        return {"error": f"Unknown parameter '{paramcd}'",
                "available": sorted(set(adlb["PARAMCD"]))[:20]}
    d = adlb.query("PARAMCD == @paramcd and AVISITN == @visit")
    if d.empty:
        return {"error": f"No records for {paramcd} at visit {visit}"}
    return {"paramcd": paramcd, "visit": visit, "n": int(d["AVAL"].count()),
            "mean": round(float(d["AVAL"].mean()), 3),
            "sd": round(float(d["AVAL"].std(ddof=1)), 3)}


TOOLS = [
    Tool("count_subjects", "Count subjects in a treatment arm, optionally by age",
         {"type": "object",
          "properties": {
              "arm": {"type": "string",
                      "enum": ["Placebo", "Drug A", "Drug B", "All"]},
              "min_age": {"type": "integer", "minimum": 0, "maximum": 120}},
          "required": ["arm"], "additionalProperties": False},
         count_subjects),
    Tool("summarise_parameter", "Mean and SD of a lab parameter at a visit",
         {"type": "object",
          "properties": {"paramcd": {"type": "string"},
                         "visit": {"type": "integer", "minimum": 0}},
          "required": ["paramcd", "visit"], "additionalProperties": False},
         summarise_parameter),
]

WarningTool design is where agent safety lives

The prompt is a suggestion. The tool surface is the actual boundary.

Never expose	Because
run_sql(query: str)	Arbitrary read and write access
run_python(code: str)	Arbitrary code execution
read_file(path: str)	Directory traversal to anything readable
send_email(to, body)	Exfiltration, and actions with consequences
delete_records(ids)	Irreversible

Design tools to be narrow, enumerated, read-only by default, and idempotent. Re-validate inside the function — the JSON schema is a hint to the model, not an enforcement boundary.

An agent whose worst possible action is “returned a wrong number, which a human checked” is a fundamentally different risk from one that can write to a database.

LangGraph

For anything beyond a single loop, LangGraph models the workflow as an explicit state graph. The production pattern is LangGraph for orchestration with LCEL chains inside individual nodes.

uv add langgraph

from typing import Annotated, TypedDict
from operator import add

from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver


class ReviewState(TypedDict):
    dataset: str
    findings: Annotated[list[str], add]     # accumulates across nodes
    checks_run: Annotated[list[str], add]
    severity: str
    needs_human: bool
    report: str


def check_structure(state: ReviewState) -> dict:
    findings = []
    df = load_dataset(state["dataset"])
    if df["USUBJID"].duplicated().any():
        findings.append("Duplicate USUBJID values found")
    if df.empty:
        findings.append("Dataset is empty")
    return {"findings": findings, "checks_run": ["structure"]}


def check_ranges(state: ReviewState) -> dict:
    findings = []
    df = load_dataset(state["dataset"])
    if "AGE" in df and ((df["AGE"] < 0) | (df["AGE"] > 120)).any():
        findings.append("AGE outside the plausible range")
    return {"findings": findings, "checks_run": ["ranges"]}


def triage(state: ReviewState) -> dict:
    n = len(state["findings"])
    severity = "none" if n == 0 else "minor" if n <= 2 else "major"
    return {"severity": severity, "needs_human": severity == "major"}


def summarise(state: ReviewState) -> dict:
    if not state["findings"]:
        return {"report": f"{state['dataset']}: no findings."}
    body = "\n".join(f"- {f}" for f in state["findings"])
    return {"report": f"{state['dataset']} ({state['severity']}):\n{body}"}


def route(state: ReviewState) -> str:
    return "escalate" if state["needs_human"] else "summarise"


graph = StateGraph(ReviewState)
graph.add_node("structure", check_structure)
graph.add_node("ranges", check_ranges)
graph.add_node("triage", triage)
graph.add_node("summarise", summarise)
graph.add_node("escalate", lambda s: {"report": "Escalated for human review."})

graph.add_edge(START, "structure")
graph.add_edge(START, "ranges")            # runs in parallel with structure
graph.add_edge("structure", "triage")
graph.add_edge("ranges", "triage")
graph.add_conditional_edges("triage", route,
                            {"summarise": "summarise", "escalate": "escalate"})
graph.add_edge("summarise", END)
graph.add_edge("escalate", END)

app = graph.compile(checkpointer=MemorySaver())

result = app.invoke({"dataset": "adsl", "findings": [], "checks_run": [],
                     "severity": "", "needs_human": False, "report": ""},
                    config={"configurable": {"thread_id": "review-1"}})
print(result["report"])

What the graph buys over the hand-rolled loop:

Feature	Benefit
Explicit nodes and edges	The execution path is readable and reviewable
Annotated[list, add] reducers	Parallel branches merge state safely
Conditional edges	Branching without the model deciding control flow
Checkpointing	Resume after failure; inspect intermediate state
Interrupts	Pause for human approval mid-run

Human in the loop

app = graph.compile(checkpointer=MemorySaver(),
                    interrupt_before=["escalate"])

config = {"configurable": {"thread_id": "review-2"}}
state = app.invoke(initial, config=config)      # pauses before "escalate"

snapshot = app.get_state(config)
print(snapshot.values["findings"])              # inspect before continuing

app.update_state(config, {"severity": "minor"}) # a human overrides
app.invoke(None, config=config)                 # resume

interrupt_before is the mechanism that makes an agent usable where the consequences matter. The pause is enforced by the framework, not requested in a prompt.

Model Context Protocol

MCP is an open standard for exposing tools, data and prompts to AI applications. Instead of writing bespoke tool code inside every application, you run a server that any MCP-capable client can use.

┌─────────────┐         MCP          ┌──────────────────┐
│   Client    │ ◀──────────────────▶ │  Your server     │
│ (assistant, │   tools / resources  │  study data,     │
│  IDE, app)  │   / prompts          │  metadata, docs  │
└─────────────┘                      └──────────────────┘

The value is reuse: one server exposing your study metadata works with every client, rather than each application reimplementing the same access layer.

uv add "mcp[cli]"

A server

from mcp.server.fastmcp import FastMCP
import pandas as pd

mcp = FastMCP("study-data")

ADSL = pd.read_parquet("data/adam/adsl.parquet")
ADLB = pd.read_parquet("data/adam/adlb.parquet")


@mcp.tool()
def count_subjects(arm: str = "All", min_age: int = 0) -> dict:
    """Count subjects in the safety population, optionally filtered.

    Args:
        arm: Treatment arm, or "All".
        min_age: Minimum age in years.
    """
    valid = set(ADSL["TRT01A"]) | {"All"}
    if arm not in valid:
        raise ValueError(f"Unknown arm '{arm}'. Valid: {sorted(valid)}")

    d = ADSL.query("SAFFL == 'Y'")
    if arm != "All":
        d = d[d["TRT01A"] == arm]
    d = d[d["AGE"] >= min_age]
    return {"arm": arm, "min_age": min_age, "n_subjects": int(len(d))}


@mcp.tool()
def parameter_summary(paramcd: str, visit: int) -> dict:
    """Mean, SD and n for a laboratory parameter at a visit."""
    available = sorted(set(ADLB["PARAMCD"]))
    if paramcd not in available:
        raise ValueError(f"Unknown parameter. Available: {available[:20]}")

    d = ADLB.query("PARAMCD == @paramcd and AVISITN == @visit and ANL01FL == 'Y'")
    if d.empty:
        raise ValueError(f"No analysis records for {paramcd} at visit {visit}")

    return {"paramcd": paramcd, "visit": visit,
            "n": int(d["AVAL"].count()),
            "mean": round(float(d["AVAL"].mean()), 4),
            "sd": round(float(d["AVAL"].std(ddof=1)), 4)}


@mcp.resource("study://metadata/{dataset}")
def dataset_metadata(dataset: str) -> str:
    """Variable-level metadata for an analysis dataset."""
    frames = {"adsl": ADSL, "adlb": ADLB}
    if dataset.lower() not in frames:
        raise ValueError(f"Unknown dataset. Available: {sorted(frames)}")

    df = frames[dataset.lower()]
    lines = [f"{dataset.upper()}: {len(df)} records, {len(df.columns)} variables", ""]
    lines += [f"  {c:<12} {str(df[c].dtype):<12} "
              f"{df[c].isna().sum():>6} missing  {df[c].nunique():>6} distinct"
              for c in df.columns]
    return "\n".join(lines)


@mcp.prompt()
def review_parameter(paramcd: str) -> str:
    """A reusable prompt for reviewing one laboratory parameter."""
    return (
        f"Review the laboratory parameter {paramcd}.\n\n"
        f"1. Use parameter_summary to get values at visits 0 and 12.\n"
        f"2. Describe the change from baseline.\n"
        f"3. Flag anything that warrants medical review.\n"
        f"Cite the numbers you used. Do not speculate beyond the data."
    )


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

Three primitive types, with different purposes:

Primitive	Semantics	Example
Tool	An action the model may invoke	count_subjects()
Resource	Data the client may read	Dataset metadata
Prompt	A reusable template the user may invoke	Parameter review

Note that tools here raise on invalid input. The MCP layer converts the exception into an error result the model can read and correct from, so you get the self-correction behaviour without hand-writing error dictionaries.

Running and connecting

mcp dev server.py                    # inspector UI for testing
mcp install server.py --name "Study ABC-101"

{
  "mcpServers": {
    "study-abc101": {
      "command": "uv",
      "args": ["--directory", "/path/to/project", "run", "python", "server.py"],
      "env": {"STUDY_DATA_PATH": "/mnt/studies/abc101"}
    }
  }
}

A client

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


async def query_server(question: str) -> str:
    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()

            tools = await session.list_tools()
            result = await session.call_tool("count_subjects", {"arm": "Placebo"})
            return result.content[0].text

ImportantAn MCP server is an access-control boundary

The server decides what any connected client can see. That makes it a good place to enforce authorisation — and a serious liability if you do not.

@mcp.tool()
def get_subject_data(usubjid: str) -> dict:
    """DANGEROUS as written — returns any subject's record to any client."""
    return ADSL[ADSL["USUBJID"] == usubjid].to_dict("records")[0]

Rules for a server touching study data:

• Expose aggregates, not subject-level records, wherever possible
• Suppress small cells — a count of 1 can be identifying
• Scope to the sites or studies the caller is authorised for
• Log every call with the caller, arguments and row counts returned
• Never expose a free-text query parameter that becomes SQL or a file path
• Run it locally over stdio rather than on a network port unless you have authenticated the transport

The same argument as Shiny authorisation: filter at the source, so no downstream consumer can reach beyond its permissions.

Guardrails

@dataclass
class Guardrails:
    max_steps: int = 10
    max_tokens: int = 100_000
    max_cost_usd: float = 1.00
    max_wall_seconds: int = 120
    tools_requiring_approval: set[str] = field(default_factory=set)
    forbidden_patterns: tuple = ("DROP ", "DELETE ", "TRUNCATE ", "--")

    def check_arguments(self, tool: str, args: dict) -> None:
        blob = json.dumps(args).upper()
        for pattern in self.forbidden_patterns:
            if pattern in blob:
                raise PermissionError(
                    f"Tool '{tool}' rejected: argument contains '{pattern.strip()}'"
                )

A layered approach, because no single control is sufficient:

Layer	Prevents
Narrow tool schemas	The agent asking for something dangerous
Argument validation	A crafted argument slipping through the schema
Step and token limits	Runaway loops
Cost budget	Runaway spend
Wall-clock timeout	Hanging on a slow tool
Approval for consequential tools	Irreversible actions without a human
Full audit log	Not knowing what happened
Read-only credentials	Everything above failing at once

The last row is the one that actually saves you. If the database credential the agent’s tools use is read-only, no combination of prompt injection and buggy validation can delete anything.

Auditing

def log_agent_run(task: str, result: AgentResult, log_path: Path) -> str:
    run_id = str(uuid.uuid4())
    record = {
        "run_id":        run_id,
        "timestamp":     datetime.now(timezone.utc).isoformat(),
        "task_hash":     hashlib.sha256(task.encode()).hexdigest(),
        "n_steps":       len(result.steps),
        "tools_used":    [s["tool"] for s in result.steps],
        "input_tokens":  result.input_tokens,
        "output_tokens": result.output_tokens,
        "halted_reason": result.halted_reason,
        "answer_hash":   hashlib.sha256(result.answer.encode()).hexdigest(),
        "steps":         result.steps,
    }
    with log_path.open("a") as f:
        f.write(json.dumps(record, default=str) + "\n")
    return run_id

Because an agent’s path varies between runs, the sequence of tool calls is the audit record. Two runs producing the same answer by different routes is normal and needs to be visible.

Agents in regulated work

An honest position.

Reasonable uses today:

• Drafting a data review query list for a human to triage
• Explaining unfamiliar legacy code to a programmer
• Exploratory investigation of a signal, with findings verified afterwards
• Generating candidate test cases for a human to accept or reject
• Answering documentation questions with citations to verify

Not reasonable:

• Deriving an analysis dataset variable
• Producing a table that goes into a submission
• Making an inclusion or exclusion decision
• Coding an adverse event term without review
• Anything where the output is not checked

The dividing line is not capability. It is that a non-deterministic component whose execution path varies between runs cannot satisfy a reproducibility requirement. An agent that helps a programmer write a validated derivation is fine; an agent inside the validated derivation is not.

That may change. The engineering practices — bounded tools, full audit trails, human approval gates, read-only credentials — are what would make it defensible when it does.

Common mistakes

Mistake	Consequence	Fix
Agent where a chain would do	Variable cost, no reproducibility	Write the steps down
No step limit	Infinite loop	max_steps
No token or cost budget	Surprise invoice	Hard limits that halt
Broad tools (run_sql, read_file)	Injection into real capability	Narrow, enumerated, read-only
Trusting the JSON schema	Model can send anything	Re-validate in the function
Tool errors raised, not returned	Agent cannot self-correct	Return the error to the model
No allowlist on tool names	Hallucinated tool dispatch	Dictionary lookup you control
Subject-level data over MCP	Re-identification risk	Aggregates; suppress small cells
No audit of the tool sequence	Cannot explain what happened	Log every step
Write credentials for agent tools	One bug is unrecoverable	Read-only

Exercise 19.1 — A bounded data review agent

Build an agent that answers questions about a study dataset using read-only tools, with a step limit, a cost budget, an approval gate for any export, and a full audit trail. Then explain what would have to be true before it could be used unsupervised.

Show solution

from __future__ import annotations

import hashlib
import json
import logging
import time
import uuid
from dataclasses import dataclass, field
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Callable

import pandas as pd

logger = logging.getLogger(__name__)


# --- Guardrails ------------------------------------------------------------

@dataclass
class Limits:
    max_steps: int = 8
    max_tokens: int = 60_000
    max_cost_usd: float = 0.50
    max_wall_seconds: int = 90
    min_cell_size: int = 5          # suppress counts below this


# --- Tools: read-only, enumerated, aggregate-only -------------------------

class StudyTools:
    """Read-only tools over analysis datasets, with small-cell suppression."""

    def __init__(self, adsl: pd.DataFrame, adlb: pd.DataFrame, limits: Limits):
        self._adsl = adsl
        self._adlb = adlb
        self.limits = limits
        self.call_log: list[dict] = []

    def _suppress(self, n: int) -> int | str:
        return n if n == 0 or n >= self.limits.min_cell_size else "<5"

    def _record(self, tool: str, args: dict, rows: int) -> None:
        self.call_log.append({"tool": tool, "args": args, "rows_touched": rows})

    # -- individual tools --

    def count_subjects(self, arm: str = "All", min_age: int = 0,
                       max_age: int = 120) -> dict:
        """Count safety-population subjects, optionally filtered by arm and age."""
        valid = sorted(set(self._adsl["TRT01A"]) | {"All"})
        if arm not in valid:
            raise ValueError(f"Unknown arm '{arm}'. Valid: {valid}")
        if not 0 <= min_age <= max_age <= 120:
            raise ValueError(f"Invalid age range {min_age}-{max_age}")

        d = self._adsl.query("SAFFL == 'Y'")
        if arm != "All":
            d = d[d["TRT01A"] == arm]
        d = d[d["AGE"].between(min_age, max_age)]

        self._record("count_subjects", {"arm": arm, "min_age": min_age}, len(d))
        return {"arm": arm, "age_range": [min_age, max_age],
                "n_subjects": self._suppress(len(d))}

    def demographics_summary(self, arm: str = "All") -> dict:
        """Mean age, sex distribution and N for an arm."""
        d = self._adsl.query("SAFFL == 'Y'")
        if arm != "All":
            if arm not in set(self._adsl["TRT01A"]):
                raise ValueError(f"Unknown arm '{arm}'")
            d = d[d["TRT01A"] == arm]
        if d.empty:
            raise ValueError(f"No safety-population subjects for arm '{arm}'")

        sex = {k: self._suppress(int(v))
               for k, v in d["SEX"].value_counts().items()}
        self._record("demographics_summary", {"arm": arm}, len(d))
        return {"arm": arm, "n": self._suppress(len(d)),
                "mean_age": round(float(d["AGE"].mean()), 1),
                "sd_age": round(float(d["AGE"].std(ddof=1)), 2),
                "sex": sex}

    def parameter_summary(self, paramcd: str, visit: int) -> dict:
        """Mean, SD and n for a lab parameter at a visit, by arm."""
        available = sorted(set(self._adlb["PARAMCD"]))
        if paramcd not in available:
            raise ValueError(f"Unknown parameter '{paramcd}'. "
                             f"Available: {available[:15]}")

        d = self._adlb.query(
            "PARAMCD == @paramcd and AVISITN == @visit and ANL01FL == 'Y'")
        if d.empty:
            raise ValueError(f"No analysis records for {paramcd} at visit {visit}")

        by_arm = (d.groupby("TRT01A")["AVAL"]
                   .agg(n="count", mean="mean", sd=lambda s: s.std(ddof=1))
                   .round(3))
        self._record("parameter_summary", {"paramcd": paramcd, "visit": visit},
                     len(d))
        return {"paramcd": paramcd, "visit": visit,
                "by_arm": {k: {"n": self._suppress(int(v["n"])),
                               "mean": v["mean"], "sd": v["sd"]}
                           for k, v in by_arm.to_dict("index").items()}}

    def list_parameters(self) -> dict:
        """Available laboratory parameters and visits."""
        return {"parameters": sorted(set(self._adlb["PARAMCD"])),
                "visits": sorted(int(v) for v in set(self._adlb["AVISITN"]))}

    def export_summary(self, filename: str) -> dict:
        """Write a summary to disk. REQUIRES APPROVAL."""
        safe = Path(filename).name              # strip any path component
        if not safe.endswith(".csv"):
            raise ValueError("Only .csv exports are permitted")
        path = Path("outputs") / safe
        path.parent.mkdir(exist_ok=True)
        pd.DataFrame(self.call_log).to_csv(path, index=False)
        return {"written": str(path), "rows": len(self.call_log)}

The agent, with every guardrail active:

@dataclass
class RunResult:
    run_id: str
    answer: str
    steps: list[dict] = field(default_factory=list)
    input_tokens: int = 0
    output_tokens: int = 0
    cost_usd: float = 0.0
    elapsed_s: float = 0.0
    halted: str | None = None


class ReviewAgent:
    SYSTEM = (
        "You answer questions about clinical study data using ONLY the provided "
        "tools.\n\n"
        "Rules:\n"
        "- Never estimate or infer a number. Call a tool for every figure.\n"
        "- Counts shown as '<5' are suppressed for privacy. Report them as "
        "'fewer than 5', never as a specific number.\n"
        "- If a tool returns an error, read it and correct your call.\n"
        "- State the tools and arguments you used for each number.\n"
        "- If the tools cannot answer, say so plainly."
    )

    APPROVAL_REQUIRED = {"export_summary"}

    def __init__(self, client, model: str, tools: StudyTools, limits: Limits,
                 in_price_per_1k: float = 0.00015,
                 out_price_per_1k: float = 0.0006):
        self.client = client
        self.model = model
        self.tools = tools
        self.limits = limits
        self.in_price = in_price_per_1k
        self.out_price = out_price_per_1k
        self.registry: dict[str, Callable] = {
            "count_subjects":       tools.count_subjects,
            "demographics_summary": tools.demographics_summary,
            "parameter_summary":    tools.parameter_summary,
            "list_parameters":      tools.list_parameters,
            "export_summary":       tools.export_summary,
        }

    def run(self, question: str,
            approve: Callable[[str, dict], bool] | None = None) -> RunResult:
        res = RunResult(run_id=str(uuid.uuid4()), answer="")
        started = time.monotonic()
        messages: list[Any] = [{"role": "user", "content": question}]

        for step in range(1, self.limits.max_steps + 1):
            elapsed = time.monotonic() - started
            if elapsed > self.limits.max_wall_seconds:
                res.halted = f"wall-clock limit ({self.limits.max_wall_seconds}s)"
                break
            if res.cost_usd > self.limits.max_cost_usd:
                res.halted = f"cost limit (${self.limits.max_cost_usd:.2f})"
                break
            if res.input_tokens + res.output_tokens > self.limits.max_tokens:
                res.halted = "token limit"
                break

            response = self.client.responses.create(
                model=self.model, instructions=self.SYSTEM, input=messages,
                tools=TOOL_SCHEMAS, temperature=0, timeout=30,
            )
            res.input_tokens += response.usage.input_tokens
            res.output_tokens += response.usage.output_tokens
            res.cost_usd = (res.input_tokens / 1000 * self.in_price
                            + res.output_tokens / 1000 * self.out_price)

            calls = [i for i in response.output if i.type == "function_call"]
            if not calls:
                res.answer = response.output_text
                break

            messages += response.output

            for call in calls:
                rec = {"step": step, "tool": call.name,
                       "arguments": call.arguments, "approved": None}

                if call.name not in self.registry:
                    out = {"error": f"Unknown tool '{call.name}'"}
                else:
                    args = json.loads(call.arguments)
                    if call.name in self.APPROVAL_REQUIRED:
                        ok = approve is not None and approve(call.name, args)
                        rec["approved"] = ok
                        out = ({"error": "Rejected by the operator"} if not ok
                               else self._safe_call(call.name, args))
                    else:
                        out = self._safe_call(call.name, args)

                rec["result"] = str(out)[:300]
                res.steps.append(rec)
                messages.append({"type": "function_call_output",
                                 "call_id": call.call_id,
                                 "output": json.dumps(out, default=str)})
        else:
            res.halted = f"step limit ({self.limits.max_steps})"

        if res.halted and not res.answer:
            res.answer = f"Halted before reaching an answer: {res.halted}."

        res.elapsed_s = time.monotonic() - started
        self._audit(question, res)
        return res

    def _safe_call(self, name: str, args: dict) -> Any:
        try:
            return self.registry[name](**args)
        except Exception as e:
            logger.warning("Tool %s failed: %s", name, e)
            return {"error": f"{type(e).__name__}: {e}"}

    def _audit(self, question: str, res: RunResult,
               path: Path = Path("logs/agent_runs.jsonl")) -> None:
        path.parent.mkdir(parents=True, exist_ok=True)
        with path.open("a") as f:
            f.write(json.dumps({
                "run_id": res.run_id,
                "timestamp": datetime.now(timezone.utc).isoformat(),
                "question_hash": hashlib.sha256(question.encode()).hexdigest(),
                "n_steps": len(res.steps),
                "tool_sequence": [s["tool"] for s in res.steps],
                "input_tokens": res.input_tokens,
                "output_tokens": res.output_tokens,
                "cost_usd": round(res.cost_usd, 6),
                "elapsed_s": round(res.elapsed_s, 2),
                "halted": res.halted,
                "answer_hash": hashlib.sha256(res.answer.encode()).hexdigest(),
                "steps": res.steps,
            }, default=str) + "\n")

def console_approval(tool: str, args: dict) -> bool:
    print(f"\nThe agent requests: {tool}({args})")
    return input("Approve? [y/N] ").strip().lower() == "y"


agent = ReviewAgent(client, MODEL, StudyTools(adsl, adlb, Limits()), Limits())

res = agent.run(
    "How does mean ALT at week 12 compare between arms, and how many "
    "subjects over 75 are in each arm?",
    approve=console_approval,
)

print(res.answer)
print(f"\n{len(res.steps)} steps, {res.elapsed_s:.1f}s, ${res.cost_usd:.4f}")
for s in res.steps:
    print(f"  {s['step']}. {s['tool']}({s['arguments']}) -> {s['result'][:70]}")

Mean ALT at week 12 was 28.4 U/L (SD 9.12, n=84) in Placebo and 24.1 U/L
(SD 8.77, n=82) in Drug A — a difference of about 4.3 U/L lower on Drug A.
[parameter_summary(paramcd="ALT", visit=12)]

Subjects over 75: 31 in Placebo, 29 in Drug A.
[count_subjects(arm="Placebo", min_age=76), count_subjects(arm="Drug A", min_age=76)]

4 steps, 6.3s, $0.0031
  1. list_parameters({}) -> {'parameters': ['ALP', 'ALT', 'AST', ...
  2. parameter_summary({"paramcd": "ALT", "visit": 12}) -> {'paramcd': 'ALT'...
  3. count_subjects({"arm": "Placebo", "min_age": 76}) -> {'n_subjects': 31}
  4. count_subjects({"arm": "Drug A", "min_age": 76}) -> {'n_subjects': 29}

Note step 1. The agent called list_parameters first to confirm ALT existed rather than guessing. That behaviour comes from the error messages — parameter_summary returns the available list on a miss, which teaches the model to check. Good tool errors shape good agent behaviour more than prompt instructions do.

---

What would have to be true before this ran unsupervised

Being direct: for anything feeding regulated output, the honest answer is that it should not, and the reasons are not fixable by better engineering.

What is already adequate:

• Tools are read-only, aggregate-only, with enumerated parameters
• Small cells suppressed at the tool layer, so the model never sees them
• Step, token, cost and wall-clock limits, all enforced in code
• Approval gate on the only tool with a side effect
• Full audit trail including the tool sequence and hashes

What is missing and achievable:

• An evaluation set of questions with known correct answers, run on every model or prompt change — the lesson 17 approach applied to the whole agent
• Numeric verification: recompute every figure in the answer directly from the data and flag any mismatch. This is the highest-value addition, because it turns “the model said 28.4” into “28.4 was independently confirmed”
• A pinned dated model version, recorded per run
• Read-only database credentials, if the tools ever query a database
• Rate limiting per user, and authorisation scoping to the caller’s studies

What is missing and not achievable:

• Reproducibility. The tool sequence varies between runs. Two runs may reach the same answer by different routes, or reach different answers. No amount of engineering makes a stochastic component deterministic.
• Completeness guarantees. The agent may not call the tool that would have revealed the important finding, and nothing detects that omission.

Therefore: usable unsupervised for exploratory triage — surfacing candidates for a human to examine, where a missed item is recovered by the existing review process. Not usable for anything where the output is the deliverable.

The productive framing is that this replaces the first thirty minutes of a data reviewer’s investigation, not the reviewer.

Recap

• An agent chooses its own next step; a chain does not. Prefer the chain
• Every agent needs step, token, cost and wall-clock limits enforced in code
• Tool design is the real security boundary — narrow, enumerated, read-only
• Return tool errors to the model so it can self-correct; good errors shape good behaviour
• LangGraph makes the execution path explicit and supports interrupts for approval
• MCP exposes tools, resources and prompts to any client — and is an access-control boundary
• Aggregate over MCP, suppress small cells, log every call
• Read-only credentials are the control that survives every other one failing
• Agents assist regulated work; they do not perform it

Course complete. Nineteen lessons: the Python language, the scientific stack, file handling and dataset comparison, clinical data in and out, statistics, machine learning, applications, testing, environments, deployment, and generative AI from prompting through to agents.

Where next:

• Clinical Programming with R — the CDISC pipeline where R’s ecosystem is stronger
• R Shiny — the framework Shiny for Python mirrors
• R Programming — the same concepts in the other language