Keyboard shortcuts

Press ← or → to navigate between chapters

Press S or / to search in the book

Press ? to show this help

Press Esc to hide this help

Chapter 10: Agentic Patterns

Data servers and API servers share one shape: the model asks, and the server answers in a single call. Those patterns cover a lot of ground, but they all assume the whole job fits inside one request and response.

Agentic patterns start where that assumption breaks: the server has to hold state across calls, pause for a human, run background work, or coordinate several steps. MCP supplies the primitives for this (tools, resources, elicitation, tasks and notifications), not the autonomy.

The key design principle is this: let the model plan, but let the server carry the hard state and the sharp edges. A model is good at reasoning about what step comes next. It is terrible at remembering what happened three calls ago unless you give it something durable to look at. Every pattern in this chapter follows from that asymmetry.

Pattern 1: Stateful Conversations

Stateless tool calls are enough for many integrations. They are not enough for workflows that stretch over time, require checkpoints, or need to remember intermediate decisions. The moment your server needs to say “given what you selected earlier,” you need state management.

What Belongs on the Server

Keep working sets, intermediate results, review queues and half-finished configuration outside the model. If the only copy of important state lives in the model’s context window, your workflow is already fragile. Context windows have hard limits, the model may summarize or drop details, and a session restart erases everything.

Server-Minted Session Handles

The 2026-07-28 protocol carries no session of its own. Every request stands alone, so cross-call state is a handle your server mints and the model passes back. That makes the handle’s design a security decision. A model-invented id gets hallucinated, reused across conversations, and on a remote server guessed. So the server mints the id, makes it unguessable, and binds it to the authenticated caller:

import secrets
from datetime import datetime, timedelta, timezone

from mcp.server.auth.middleware.auth_context import get_access_token
from mcp.server.mcpserver import MCPServer
from mcp.server.mcpserver.exceptions import ToolError

mcp = MCPServer("Stateful Workflow Server")

SESSION_TTL = timedelta(minutes=30)

# In production, use Redis or a database. This is for illustration.
_sessions: dict[str, dict] = {}


def caller() -> str:
    token = get_access_token()
    if token is None:
        raise ToolError("This server requires an authenticated caller.")
    return f"{token.client_id}:{token.subject}"


def now() -> datetime:
    return datetime.now(timezone.utc)


@mcp.tool()
def start_session() -> dict:
    """Start a workflow session. Pass the returned session_id to later calls."""
    session_id = secrets.token_urlsafe(24)
    _sessions[session_id] = {
        "owner": caller(),
        "expires_at": now() + SESSION_TTL,
        "selections": [],
        "intermediate_results": {},
    }
    return {"session_id": session_id,
            "expires_at": _sessions[session_id]["expires_at"].isoformat()}


def get_session(session_id: str) -> dict:
    """Refuse unknown, expired or foreign ids. Never create a session here."""
    session = _sessions.get(session_id)
    if session is None or session["owner"] != caller():
        raise ToolError(f"Unknown session {session_id}. "
                        "Call start_session to begin a new one.")
    if now() >= session["expires_at"]:
        del _sessions[session_id]
        raise ToolError(
            f"Session {session_id} expired at "
            f"{session['expires_at']:%H:%M} UTC and its selections are gone. "
            "Call start_session and re-select repositories."
        )
    session["expires_at"] = now() + SESSION_TTL  # extend on activity
    return session


@mcp.tool()
def select_repository(session_id: str, repo_name: str) -> dict:
    """Add a repository to the session's working set."""
    session = get_session(session_id)
    if repo_name not in session["selections"]:
        session["selections"].append(repo_name)
    return {"working_set": session["selections"]}


@mcp.tool()
def run_analysis(session_id: str) -> dict:
    """Run analysis across all selected repositories."""
    session = get_session(session_id)
    if not session["selections"]:
        raise ToolError("No repositories selected. Call select_repository first.")
    result = {"repos_analyzed": len(session["selections"]), "issues_found": 42}
    session["intermediate_results"]["last_analysis"] = result
    return result


@mcp.tool()
def get_session_state(session_id: str) -> dict:
    """Current working set and results, without side effects."""
    session = get_session(session_id)
    return {k: v for k, v in session.items() if k != "owner"}

The owner check returns the same “unknown session” error as a missing id, so a caller probing for other people’s ids learns nothing. The inspection tool (get_session_state) is not optional. When a model drifts or a user asks “what have I selected so far?”, the model needs a way to check reality rather than guessing from its context window.

Session Lifecycle

Sessions need expiry, or your server leaks storage indefinitely. Set a TTL at creation (30 minutes for interactive work), extend it on every call, offer a close_session tool, and run a reaper that deletes expired sessions.

The failure mode here is not session expiry itself but silent expiry. The tempting get_session creates a fresh, empty session for any id it doesn’t recognize. An expired or mistyped id then quietly becomes a blank working set, and run_analysis reports on nothing as if that were the answer. This is why get_session refuses unknown ids and names the expiry time.

Pattern 2: Human in the Loop

Some actions should not complete without human confirmation. A model deciding to delete a production database or send a bulk email needs a gate. There are three ways to build one, and they differ in who enforces them.

Host-Level Approvals

The simplest gate is the one you do not build: tool annotations. When you annotate a tool with destructiveHint: true, well-behaved clients present an approval dialog before executing the call.

@mcp.tool(
    annotations={
        "title": "Deploy to Production",
        "destructiveHint": True,
        "readOnlyHint": False,
    }
)
def deploy_to_production(service_name: str, version: str) -> str:
    """Deploy a service version to the production environment."""
    # Client will prompt user for approval before this executes
    return f"Deployed {service_name} v{version} to production."

The limitation is that annotations are hints, not guarantees. Not every client implements approval UIs, and the spec says clients MAY ignore hints. If your safety requirement is hard — “never deploy without a human yes” — annotations alone are insufficient.

Elicitation: Structured User Input Mid-Execution

Elicitation lets a server request structured input from the user through the client before the tool acts. Unlike annotations, which are pre-execution gates the client may ignore, elicitation is enforced by your server, and your code decides what happens with the answer. In the SDK, a resolver asks the question and the tool body receives the answer:

from typing import Annotated

from pydantic import BaseModel

from mcp.server.elicitation import ElicitationResult
from mcp.server.mcpserver import Elicit, Resolve


class DeploymentApproval(BaseModel):
    approved: bool
    reason: str


def ask_approval(service: str, version: str) -> Elicit[DeploymentApproval]:
    # May run on every round: checks only, no side effects (see below)
    return Elicit(f"Approve deployment of {service} v{version} to production?",
                  DeploymentApproval)


@mcp.tool()
async def deploy_with_review(
    service: str, version: str,
    approval: Annotated[ElicitationResult[DeploymentApproval], Resolve(ask_approval)],
) -> str:
    """Deploy a service after collecting explicit human approval."""
    if approval.action != "accept" or not approval.data.approved:
        return f"Deployment of {service} v{version} not approved; nothing changed."
    await deploy(service, version)  # your deployment call
    return f"Deployed {service} v{version}. Reason: {approval.data.reason}"

The approval parameter never appears in the tool’s input schema; the framework fills it. A user can accept, decline or cancel, and your server must handle all three. The schema is limited to flat objects with primitive properties.

Your Handler May Run Twice

Under 2026-07-28, an elicitation is a retry. The server doesn’t hold the call open while the user decides. It returns an InputRequiredResult, and the client re-sends the original tools/call with the answer attached in inputResponses. The resolver hides the mechanics, but not the consequence: a traced call runs ask_approval, the elicitation, ask_approval again, then the body once. Calling ctx.elicit() inside the body instead fails on a 2026-07-28 connection, because there is no open request to send the question on.

Suppose ask_approval wrote an audit row, “deployment requested”, before returning Elicit. One approval would produce two rows. If it reserved a deploy slot, it would reserve two. Keep resolvers to checks, put side effects in the body, or make them idempotent: key the audit row on the request so the second insert is a no-op.

Confirmation Tools as Fallback

When the client doesn’t support elicitation, build confirmation into your tool surface:

_pending_actions: dict[str, dict] = {}


@mcp.tool()
def request_deployment(service: str, version: str) -> str:
    """Stage a deployment. A human must approve it before confirm_deployment."""
    confirmation_id = secrets.token_urlsafe(8)
    _pending_actions[confirmation_id] = {
        "owner": caller(),
        "service": service,
        "version": version,
        "expires_at": now() + timedelta(minutes=10),
    }
    return (f"Deployment staged: {service} v{version}. "
            f"Confirmation ID: {confirmation_id}. Expires in 10 minutes.")


@mcp.tool()
def confirm_deployment(confirmation_id: str) -> str:
    """Execute a previously staged deployment."""
    action = _pending_actions.get(confirmation_id)
    if action is None or action["owner"] != caller():
        raise ToolError(f"No pending deployment with ID {confirmation_id}.")
    del _pending_actions[confirmation_id]  # only the owner consumes it
    if now() >= action["expires_at"]:
        raise ToolError(f"Deployment {confirmation_id} expired unconfirmed. "
                        "Stage it again with request_deployment.")
    return f"Deployed {action['service']} v{action['version']} to production."

This works with every client, but read it for what it enforces. The staging step is a gate only if confirmation needs something the model cannot produce: an elicited yes, a code the user reads out, or a client approval. Otherwise the model can stage and confirm in the same turn.

Design for Timeout and Abandonment

Human-in-the-loop systems fail as often from silence as from rejection. Apply TTLs to pending actions, as request_deployment does. When a confirmation expires, say what timed out and how to restart.

Pattern 3: Long-Running Jobs

A batch export or a large sync doesn’t fit a request-response shape. The call blocks, the transport times out, and the model can do nothing else while it waits.

Job Handles

The portable approach is to return a job handle immediately and expose a status tool:

import asyncio
import uuid

_jobs: dict[str, dict] = {}
_running: set[asyncio.Task] = set()  # keep references so tasks aren't collected


async def _run_export(job_id: str, dataset: str, fmt: str) -> None:
    job = _jobs[job_id]
    try:
        for i in range(10):
            await asyncio.sleep(1)  # stands in for one chunk of real work
            job["progress"] = (i + 1) * 10
        job["result"] = f"/exports/{dataset}.{fmt}"
        job["status"] = "completed"
    except Exception as e:
        job["status"], job["error"] = "failed", str(e)


@mcp.tool()
async def start_export(dataset: str, format: str = "csv") -> dict:
    """Start a data export. Returns a job ID for get_job_status."""
    job_id = uuid.uuid4().hex[:12]
    _jobs[job_id] = {"owner": caller(), "status": "running", "progress": 0}
    task = asyncio.create_task(_run_export(job_id, dataset, format))
    _running.add(task)
    task.add_done_callback(_running.discard)
    return {"job_id": job_id, "status": "running", "poll_after_seconds": 5}


@mcp.tool()
def get_job_status(job_id: str) -> dict:
    """Status, progress and (when done) result of an export job."""
    job = _jobs.get(job_id)
    if job is None or job["owner"] != caller():
        raise ToolError(f"No job with ID {job_id}.")
    status = {k: v for k, v in job.items() if k != "owner"}
    if job["status"] == "running":
        status["poll_after_seconds"] = 5 if job["progress"] < 50 else 2
    return status

poll_after_seconds is the field that makes this safe. Without it, the model picks its own interval, and some models pick “immediately, again”. Tell it when to come back.

Tasks

Teams kept reinventing job handles with slightly different status names and polling conventions, so the protocol standardized them. In 2026-07-28, Tasks are an official extension, io.modelcontextprotocol/tasks. A server may answer any tools/call with a task handle instead of a result. The client polls tasks/get until the task completes, fails or is cancelled, and it answers mid-task questions through tasks/update. Field names and states are in the Tasks extension spec.

The one design rule: keep fast tools synchronous, and return a task only for work that runs long. Use Tasks when your target clients support the extension. Otherwise use job handles, but model their state the Tasks way (create, progress, result, cancel, expire), so that migrating later changes the transport and not the design.

Pattern 4: Multi-Step Workflows

Multi-step workflows are where models look smart and where systems break. The model drives a series of tool calls toward a composite goal, and each step depends on the ones before it. A workflow server decomposes the work into visible steps, stores checkpoints outside the model, and makes recovery explicit.

A Concrete Example: Deployment Report

Suppose the model is preparing a weekly deployment report:

  1. Fetch recent deploys from the CI/CD API.
  2. Pull incident summaries from the monitoring system.
  3. Match incidents to deploy windows.
  4. Produce a draft report.
  5. Ask for human confirmation before sending.

If step 4 fails, the system should not have to restart from step 1. That is the value of checkpoints.

@mcp.tool()
def start_deployment_report(session_id: str) -> str:
    """Begin a deployment report. Steps, in order: fetch_deploys,
    fetch_incidents, correlate_data, generate_draft, approve_and_send."""
    session = get_session(session_id)
    session["workflow"] = {
        "type": "deployment_report",
        "step": "initialized",
        "deploys": None,
        "incidents": None,
        "correlations": None,
        "draft": None,
    }
    return "Deployment report workflow started. Next: call fetch_deploys."


@mcp.tool()
def fetch_deploys(session_id: str, days: int = 7) -> str:
    """Fetch recent deployments for the report."""
    session = get_session(session_id)
    wf = session.get("workflow")
    if not wf or wf["type"] != "deployment_report":
        raise ToolError("No active deployment report workflow. "
                        "Call start_deployment_report first.")

    # Fetch from CI/CD API...
    deploys = [{"service": "api", "version": "2.3.1",
                "timestamp": "2026-03-28T14:00:00Z"}]
    wf["deploys"] = deploys
    wf["step"] = "deploys_fetched"
    return f"Fetched {len(deploys)} deploys. Next: call fetch_incidents."

Each tool checks the workflow state, refuses to execute out of order, and tells the model what to do next. If the model calls correlate_data before fetch_incidents, it gets a clear error instead of a silent failure.

Who Drives the Workflow?

The example above hands the model six tools and a required order. Before you copy it, ask the question it skips: why isn’t this one generate_deployment_report tool?

The rule: expose steps to the model only when choosing the next step needs judgment. If the model has to decide whether to skip incidents for a quiet week, retry one source, or branch on what a step returned, it needs the steps. If the order is fixed and every run makes the same five calls in the same sequence, the model adds nothing but failure points. Run the sequence server-side.

Count what model-driven steps cost. Six tools add six descriptions to every request. Each step is a round trip through the model, with one more chance to call things out of order, forget the session id, or stop halfway. The checkpoint and ordering code above exists only because the model holds the sequence. A server-driven version needs none of it:

  • generate_deployment_report(days) runs the fetches concurrently, correlates, and drafts.
  • It returns a task handle (or a job handle) if the run is long.
  • It elicits approval once, at the end, before sending. That’s the only point where a human changes the outcome.

This is Chapter 9’s aggregator rule, stretched over time: compose around the question the user asked. Model-driven steps are the exception you choose on purpose, for workflows whose shape depends on what the model sees along the way. When you do choose them, the checkpoint discipline in this section is mandatory.

Recovery Design

When a step fails mid-workflow, the model needs to know:

  • What succeeded — which steps completed and what data is checkpointed.
  • What failed — the specific error, not a generic “something went wrong.”
  • What to do next — retry the failed step, skip it, or abort.

Expose a get_workflow_status tool that returns the current step, checkpointed data, and any error information. The model can use this to decide whether to retry, ask the user for help, or report the failure.

Guarding Against Infinite Loops

Models sometimes get stuck in retry loops — calling the same failing tool repeatedly. Your server should track retry counts per step and fail hard after a reasonable limit (three attempts is typical). Return a message that explicitly tells the model to stop retrying and escalate to the user.

Failure Modes

Agentic patterns introduce failure categories that simpler servers do not face. Here are the ones that cause production incidents:

State drift. The model’s picture of session state diverges from the server’s, through swallowed errors, silent expiry or a context window that dropped earlier state. Mitigation: always expose a state inspection tool, and include current state summaries in tool responses.

Session ids the model can forge. A session keyed on an id the model invents, or one that isn’t bound to a caller, lets one conversation read or overwrite another’s working set. Mint ids server-side and check the owner on every lookup.

Duplicated side effects on elicitation. Work done in a resolver, before it returns Elicit, runs again when the client retries with the answer. You get two audit rows, two reservations, or two emails for one approval.

Approval deadlocks. A workflow blocks on human approval and the human never responds. Without timeouts, the workflow hangs indefinitely, potentially holding resources. Mitigation: TTLs on all pending actions, with clear timeout messages that explain what expired and how to restart.

Polling storms. A model checking job status in a tight loop generates unnecessary load, polling every second for a job that takes ten minutes. Mitigation: return poll_after_seconds from every status response, as get_job_status does, and rate-limit the status tool.

Checkpoint corruption. Partial writes to session or workflow state leave the data in an inconsistent state. Step 3 completes but the checkpoint write fails, so the workflow thinks it is still on step 2. Mitigation: atomic state updates, and a recovery tool that can detect and resolve inconsistencies.

Capability assumption. The server uses elicitation or Tasks without checking whether the client supports them. The call fails at runtime, and the workflow breaks. Mitigation: read the capabilities the client declares on each request, and keep a fallback path for clients that lack them.

Runaway workflows. A model chains tool calls without limit, optimistically retrying or exploring, and runs up costs or downstream rate limits. Mitigation: per-session call budgets, explicit “workflow complete” signals, and hard limits on workflow step counts.

Implementation Checklist

Before shipping an agentic MCP server:

  1. Sessions, workflows, jobs and pending actions live in durable storage with TTLs. A server restart loses nothing that hasn’t expired.
  2. get_session has no create branch. A test calls it with a made-up id and another caller’s id and gets the same “unknown session” refusal for both. An expired id gets the error that names its expiry time.
  3. Every tool that elicits is tested for a double run. Call it, answer the elicitation, and assert each side effect happened once.
  4. Elicitation and Tasks are used only when the client declares the capability, with a tested fallback path for clients that don’t.
  5. Confirmation tools require something the model can’t produce. Otherwise, label them as speed bumps in the design doc, not as gates.
  6. Every job status response carries poll_after_seconds.
  7. Each fixed-order workflow is one server-driven tool. Every model-driven step list has a written reason why the next step needs judgment.
  8. Every workflow has a status tool, a per-step retry cap and a per-session call budget.

Sampling (deprecated). Sampling let a server ask the client’s model for a completion, for summarizing, classifying or drafting mid-tool. 2026-07-28 deprecates it, and it keeps working until at least the first revision released on or after 2027-07-28. If you meet it in an existing server, it is a server-initiated model call gated by the client. New servers should call a model API directly.

What Comes Next

This chapter’s rule, “let the model plan, but let the server carry the hard state and the sharp edges,” now has code behind it: server-minted sessions, gates the model can’t pass alone, job handles that say when to poll, and a test for when the model should drive at all. Chapter 11 assembles these patterns into two projects: a personal task manager, and an operations copilot that puts server-held incident state, an elicitation-gated restart and a long-running runbook job together.