Structured Input/Output with Pydantic & Zod Schemas

Type-safe schemas do two jobs: they generate the inputSchema the model sees in tools/list (so it knows exactly what arguments are valid before calling), and they reject malformed calls before your business logic ever runs. Since mid-2025, the MCP spec also supports structured tool output, letting servers return typed JSON rather than only prose, which matters whenever a tool’s result feeds another tool call or a UI. This lesson also builds a complete error taxonomy, since a validated input schema is only half of a robust tool contract.

Input Validation with Pydantic

from pydantic import BaseModel, Field, field_validator, model_validator

class SearchAccountsInput(BaseModel):
    query: str = Field(..., min_length=2, max_length=200)
    tier: str | None = Field(None, description="Filter by tier: startup, growth, enterprise")
    min_arr: float = Field(0, ge=0)
    max_arr: float | None = Field(None, ge=0)

    @field_validator("tier")
    @classmethod
    def validate_tier(cls, v):
        allowed = {"startup", "growth", "enterprise"}
        if v is not None and v not in allowed:
            raise ValueError(f"tier must be one of {allowed}")
        return v

    @model_validator(mode="after")
    def validate_arr_range(self):
        # Cross-field validation: a single field_validator cannot see both bounds
        if self.max_arr is not None and self.max_arr < self.min_arr:
            raise ValueError("max_arr must be greater than or equal to min_arr")
        return self

@mcp.tool(name="crm_search_accounts")
async def search_accounts(input: SearchAccountsInput) -> dict:
    """Search CRM accounts by name/keyword, optionally filtered by tier and ARR range."""
    # By the time this line runs, every field, and the relationship between
    # min_arr and max_arr, is already guaranteed valid.
    results = await crm_client.search(input.query, tier=input.tier, min_arr=input.min_arr, max_arr=input.max_arr)
    return {"count": len(results), "accounts": [r.model_dump() for r in results]}

The model_validator(mode="after") is worth calling out specifically: single-field validators (field_validator) cannot see other fields, so any rule that depends on the relationship between fields, here, that max_arr cannot be smaller than min_arr, needs a model-level validator that runs after all individual fields have already passed their own checks. This is a common gap in early tool schemas: teams validate each field in isolation and miss the cross-field cases entirely, letting a nonsensical combination (a max below a min) through to the downstream query, where it either silently returns zero results or, worse, gets misinterpreted by the downstream system.

Equivalent Validation in TypeScript with Zod

import { z } from "zod";

const SearchAccountsInput = z.object({
  query: z.string().min(2).max(200),
  tier: z.enum(["startup", "growth", "enterprise"]).optional(),
  minArr: z.number().nonnegative().default(0),
  maxArr: z.number().nonnegative().optional(),
}).refine(
  (data) => data.maxArr === undefined || data.maxArr >= data.minArr,
  { message: "maxArr must be greater than or equal to minArr", path: ["maxArr"] }
);

server.registerTool(
  "crm_search_accounts",
  {
    description: "Search CRM accounts by name/keyword, optionally filtered by tier and ARR range.",
    inputSchema: SearchAccountsInput.shape,
  },
  async ({ query, tier, minArr, maxArr }) => {
    const results = await crmClient.search(query, { tier, minArr, maxArr });
    return {
      content: [{ type: "text", text: JSON.stringify({ count: results.length, accounts: results }) }],
      structuredContent: { count: results.length, accounts: results },
    };
  }
);

Zod’s .refine() plays exactly the role of Pydantic’s model_validator, cross-field validation expressed once, at the schema level, rather than scattered as ad hoc if checks inside the handler function.

A Complete Error Taxonomy

Treat tool failures as three distinct classes, each requiring different handling, rather than collapsing everything into a single try/except:

flowchart TD
    Call["tools/call crm_search_accounts"] --> Validate{"Pydantic/Zod\nvalidation"}
    Validate -->|invalid| RejectErr["1. Protocol-level error\n(returned before tool body runs,\ncaller should fix the call, not retry as-is)"]
    Validate -->|valid| Execute["Tool function executes"]
    Execute --> Downstream{"Downstream system\nresult"}
    Downstream -->|full success| Success["Structured result returned"]
    Downstream -->|total failure| ExecErr["2. Structured execution error\n(isError: true, retryable or not,\ndepending on cause)"]
    Downstream -->|partial success| Partial["3. Partial-success result\n(succeeded items + failed items\nwith per-item reasons)"]

    style Call fill:#EEF0F7,stroke:#6366F1,color:#0F172A
    style Validate fill:#fff7ed,stroke:#f59e0b,color:#0F172A
    style RejectErr fill:#fef2f2,stroke:#dc2626,color:#0F172A
    style Execute fill:#f0fdf9,stroke:#0D9488,color:#0F172A
    style Success fill:#f0fdf9,stroke:#0D9488,color:#0F172A
    style ExecErr fill:#fef2f2,stroke:#dc2626,color:#0F172A
    style Partial fill:#fff7ed,stroke:#f59e0b,color:#0F172A

Class 1, validation errors (bad input shape) are rejected by the schema layer before your code runs and should never require a try/except in the tool body, the caller needs to fix the arguments, retrying identically will fail identically. Class 2, execution errors (valid input, but the downstream system failed entirely, timed out, or returned nothing) happen inside your tool logic and should be returned as a structured, informative error, distinguishing retryable failures (a timeout, a 503) from non-retryable ones (a 404 for a genuinely nonexistent record) matters here, since a client’s retry policy should treat them differently.

@mcp.tool(name="crm_get_account")
async def get_account(account_id: str) -> dict:
    """Fetch a single CRM account by ID."""
    try:
        account = await crm_client.get_account(account_id)
    except CRMTimeoutError:
        return {"isError": True, "retryable": True, "message": "CRM system timed out, safe to retry"}
    if account is None:
        # Non-retryable: valid input, but no such account exists. Retrying
        # this exact call will never succeed, so retryable is False.
        return {"isError": True, "retryable": False, "message": f"No account found with id '{account_id}'"}
    return account.model_dump()

Class 3, partial-success results apply specifically to batch or multi-item operations, where some items succeed and others fail independently. Collapsing this into either a full success (silently dropping failures) or a full failure (discarding successes) loses information the caller needs to act correctly.

class BulkUpdateInput(BaseModel):
    account_ids: list[str] = Field(..., min_length=1, max_length=100)
    new_tier: str

@mcp.tool(name="crm_bulk_update_tier")
async def bulk_update_tier(input: BulkUpdateInput) -> dict:
    """Update tier for multiple accounts. Reports per-account success/failure explicitly."""
    succeeded, failed = [], []
    for account_id in input.account_ids:
        try:
            await crm_client.update_tier(account_id, input.new_tier)
            succeeded.append(account_id)
        except AccountNotFoundError:
            failed.append({"account_id": account_id, "reason": "not_found"})
        except CRMValidationError as e:
            failed.append({"account_id": account_id, "reason": str(e)})

    return {
        "succeeded_count": len(succeeded),
        "failed_count": len(failed),
        "succeeded": succeeded,
        "failed": failed,  # Caller can retry just these, or surface them to a human
    }

This closes out Module 2, you now have a server with validated tools, resources, and prompts, and a principled way to report every class of outcome back to the caller. Module 3 moves to transports and deployment: how this server actually gets exposed to remote clients rather than just running locally over stdio.