# How to Use Pydantic Models in MCP Python Servers: A Complete FastMCP Guide

> Learn how to use Pydantic models in MCP Python servers with FastMCP. Automatically derive tool schemas and validate arguments, eliminating manual sanitization.

- Repository: [Composio/awesome-codex-skills](https://github.com/composiohq/awesome-codex-skills)
- Tags: how-to-guide
- Published: 2026-04-26

---

**FastMCP automatically derives tool schemas from Pydantic v2 models and validates incoming arguments before execution, eliminating manual input sanitization in MCP Python servers.**

The Model Context Protocol (MCP) Python ecosystem centers on the FastMCP framework, which treats Pydantic models as first-class citizens for tool definition. When building MCP Python servers according to the ComposioHQ/awesome-codex-skills repository, you declare tool parameters as Pydantic `BaseModel` instances to leverage automatic schema generation, strict type validation, and self-documenting APIs without boilerplate validation code.

## Why FastMCP Uses Pydantic v2

FastMCP integrates tightly with Pydantic v2 (not the deprecated v1) to provide runtime validation and schema generation. According to [`mcp-builder/reference/python_mcp_server.md`](https://github.com/ComposioHQ/awesome-codex-skills/blob/main/mcp-builder/reference/python_mcp_server.md), the framework inspects function signatures at registration time; when it detects a `BaseModel` type hint on the first parameter, it extracts field types, `Field` metadata, and custom validators to build the tool's `inputSchema`.

Key advantages include:

- **Declarative validation**: Use `field_validator` for business logic instead of manual if-statements
- **JSON Schema export**: Clients receive OpenAPI-compatible schemas via `tools/list`
- **Automatic error handling**: Validation failures return structured MCP errors with `isError: true` rather than uncaught exceptions

## Initializing the FastMCP Server

Start by creating a FastMCP instance following the `{service}_mcp` naming convention outlined in [`mcp-builder/reference/mcp_best_practices.md`](https://github.com/ComposioHQ/awesome-codex-skills/blob/main/mcp-builder/reference/mcp_best_practices.md).

```python
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("example_mcp")  # follows {service}_mcp pattern

```

The server instance maintains the tool registry and orchestrates validation. When you call `mcp.run()`, FastMCP activates the stdio transport by default, though the server can also run over HTTP or SSE as implemented in [`mcp-builder/scripts/connections.py`](https://github.com/ComposioHQ/awesome-codex-skills/blob/main/mcp-builder/scripts/connections.py).

## Defining Tool Inputs with Pydantic

Create Pydantic models that describe your tool's contract. The following example demonstrates `ConfigDict` for global settings, `Field` constraints, and `field_validator` for custom logic.

```python
from enum import Enum
from typing import Optional
from pydantic import BaseModel, Field, field_validator, ConfigDict

class ResponseFormat(str, Enum):
    MARKDOWN = "markdown"
    JSON = "json"

class UserSearchInput(BaseModel):
    """Input model for the example_search_users tool."""
    model_config = ConfigDict(
        str_strip_whitespace=True,   # auto-strip strings

        validate_assignment=True,    # re-validate on assignment

        extra="forbid"               # reject unknown fields

    )

    query: str = Field(
        ...,
        description="Search string (name, email, or team filter)",
        min_length=2,
        max_length=200,
    )
    limit: Optional[int] = Field(
        default=20,
        description="Maximum results (1-100)",
        ge=1,
        le=100,
    )
    offset: Optional[int] = Field(
        default=0,
        description="Pagination offset",
        ge=0,
    )
    response_format: ResponseFormat = Field(
        default=ResponseFormat.MARKDOWN,
        description="Desired output format",
    )

    @field_validator("query")
    @classmethod
    def _not_empty(cls, v: str) -> str:
        """Additional business-logic validation."""
        if not v.strip():
            raise ValueError("query cannot be empty or whitespace")
        return v.strip()

```

The `ConfigDict` centralizes model behavior: `extra="forbid"` prevents clients from sending hallucinated parameters, while `str_strip_whitespace` normalizes input automatically.

## Registering Tools with @mcp.tool

Decorate your coroutine with `@mcp.tool` to register it. FastMCP inspects the first parameter's type hint; if it is a `BaseModel`, the framework stores the generated schema as the tool's `inputSchema`.

```python
@mcp.tool(
    name="example_search_users",
    annotations={
        "title": "Search Users (Example Service)",
        "readOnlyHint": True,
        "destructiveHint": False,
        "idempotentHint": True,
        "openWorldHint": True,
    },
)
async def example_search_users(params: UserSearchInput) -> str:
    """
    Search for users in the Example service.
    
    The `params` argument is a pre-validated UserSearchInput model.
    """
    # Access validated fields directly: params.query, params.limit, etc.

    return f"Searching for {params.query} with limit {params.limit}"

```

As noted in [`python_mcp_server.md`](https://github.com/ComposioHQ/awesome-codex-skills/blob/main/python_mcp_server.md), the decorator extracts the Pydantic model's JSON representation and exposes it to clients via the MCP protocol's `tools/list` endpoint.

## Complete Server Implementation

Below is the consolidated, runnable example from the ComposioHQ/awesome-codex-skills analysis, combining validation, HTTP error handling, and response formatting.

```python

# example_server.py

from enum import Enum
from typing import Optional

import httpx
from pydantic import BaseModel, Field, field_validator, ConfigDict

from mcp.server.fastmcp import FastMCP

# -------------------------------------------------

# 1️⃣ Initialise the MCP server

# -------------------------------------------------

mcp = FastMCP("example_mcp")

# -------------------------------------------------

# 2️⃣ Define Pydantic models

# -------------------------------------------------

class ResponseFormat(str, Enum):
    MARKDOWN = "markdown"
    JSON = "json"

class UserSearchInput(BaseModel):
    model_config = ConfigDict(
        str_strip_whitespace=True,
        validate_assignment=True,
        extra="forbid"
    )

    query: str = Field(
        ...,
        description="Search string (name, email, or team filter)",
        min_length=2,
        max_length=200,
    )
    limit: Optional[int] = Field(
        default=20,
        description="Maximum results (1-100)",
        ge=1,
        le=100,
    )
    offset: Optional[int] = Field(
        default=0,
        description="Pagination offset",
        ge=0,
    )
    response_format: ResponseFormat = Field(
        default=ResponseFormat.MARKDOWN,
        description="Desired output format",
    )

    @field_validator("query")
    @classmethod
    def _not_empty(cls, v: str) -> str:
        if not v.strip():
            raise ValueError("query cannot be empty or whitespace")
        return v.strip()

# -------------------------------------------------

# 3️⃣ Shared HTTP helper

# -------------------------------------------------

API_BASE_URL = "https://api.example.com/v1"
CHARACTER_LIMIT = 25_000

async def _make_api_request(endpoint: str, method: str = "GET", **kwargs) -> dict:
    """Reusable async HTTP client."""
    async with httpx.AsyncClient() as client:
        resp = await client.request(
            method,
            f"{API_BASE_URL}/{endpoint}",
            timeout=30.0,
            **kwargs,
        )
        resp.raise_for_status()
        return resp.json()

def _handle_api_error(exc: Exception) -> str:
    """Standardised error messages for tool callers."""
    if isinstance(exc, httpx.HTTPStatusError):
        status = exc.response.status_code
        if status == 404:
            return "Error: Resource not found."
        if status == 403:
            return "Error: Permission denied."
        if status == 429:
            return "Error: Rate limit exceeded."
        return f"Error: API request failed (status {status})."
    if isinstance(exc, httpx.TimeoutException):
        return "Error: Request timed out."
    return f"Error: Unexpected problem – {type(exc).__name__}"

# -------------------------------------------------

# 4️⃣ Register tool with Pydantic model

# -------------------------------------------------

@mcp.tool(
    name="example_search_users",
    annotations={
        "title": "Search Users (Example Service)",
        "readOnlyHint": True,
        "destructiveHint": False,
        "idempotentHint": True,
        "openWorldHint": True,
    },
)
async def example_search_users(params: UserSearchInput) -> str:
    """
    Search for users in the Example service.
    
    Validation errors are automatically turned into MCP tool errors.
    """
    try:
        data = await _make_api_request(
            "users/search",
            params={"q": params.query, "limit": params.limit, "offset": params.offset},
        )
        users = data.get("users", [])
        total = data.get("total", 0)

        if not users:
            return f"No users found matching '{params.query}'."

        if params.response_format == ResponseFormat.MARKDOWN:
            lines = [
                f"# User Search Results for `{params.query}`",

                "",
                f"Found **{total}** users (showing {len(users)}).",
                "",
            ]
            for u in users:
                lines.append(f"## {u['name']} ({u['id']})")

                lines.append(f"- Email: {u['email']}")
                if u.get("team"):
                    lines.append(f"- Team: {u['team']}")
                lines.append("")
            result = "\n".join(lines)
        else:
            import json
            result = json.dumps({"total": total, "count": len(users), "users": users}, indent=2)

        # Enforce CHARACTER_LIMIT

        if len(result) > CHARACTER_LIMIT:
            result = result[:CHARACTER_LIMIT] + "\n\n*Response truncated – use `offset`/`limit` for more.*"

        return result

    except Exception as exc:
        return _handle_api_error(exc)

# -------------------------------------------------

# 5️⃣ Run the server (default stdio transport)

# -------------------------------------------------

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

```

## How Validation Works Under the Hood

When a client calls `example_search_users`, FastMCP executes the following validation flow according to [`python_mcp_server.md`](https://github.com/ComposioHQ/awesome-codex-skills/blob/main/python_mcp_server.md):

1. **Deserialization**: Incoming JSON arguments are parsed from the MCP request
2. **Model instantiation**: FastMCP constructs `UserSearchInput(**arguments)`
3. **Validation**: Pydantic runs type checks, `Field` constraints, and `field_validator` methods
4. **Error translation**: If validation fails, FastMCP returns an MCP tool error before your function executes
5. **Execution**: Only fully-validated model instances reach the coroutine body

This guarantees that `params.query`, `params.limit`, and other fields always contain sanitized, type-correct data, while validation errors surface as clean MCP responses rather than 500 errors.

## Testing Pydantic-Based Tools

Validate your Pydantic-based tools using the evaluation patterns in [`mcp-builder/scripts/evaluation.py`](https://github.com/ComposioHQ/awesome-codex-skills/blob/main/mcp-builder/scripts/evaluation.py). This script demonstrates how MCP tools are discovered and invoked programmatically, allowing you to unit test validation logic without spinning up a full client. The [`mcp-builder/scripts/connections.py`](https://github.com/ComposioHQ/awesome-codex-skills/blob/main/mcp-builder/scripts/connections.py) file further shows how the same FastMCP server runs across stdio, HTTP, or SSE transports regardless of your Pydantic model complexity.

## Summary

- **FastMCP** automatically converts Pydantic `BaseModel` definitions into MCP `inputSchema` JSON for client discovery
- Use **ConfigDict** to set global validation rules like `extra="forbid"` and `str_strip_whitespace`
- Declare **field_validator** methods for business-logic checks beyond standard type constraints
- Pass the Pydantic model as the **first parameter** in `@mcp.tool` decorated functions to enable automatic validation
- Validation errors become **structured MCP responses** with `isError: true` rather than uncaught exceptions
- Follow the **{service}_mcp** naming convention when initializing FastMCP instances as documented in [`mcp_best_practices.md`](https://github.com/ComposioHQ/awesome-codex-skills/blob/main/mcp_best_practices.md)

## Frequently Asked Questions

### How does FastMCP handle Pydantic validation errors?

When Pydantic validation fails, FastMCP catches the `ValidationError` and returns it as a standard MCP tool error with `isError: true`. This prevents your tool function from executing with invalid data and provides clear error messages to the client without requiring try/except blocks in your business logic to catch validation issues.

### Can I use Pydantic v1 with MCP Python servers?

No. The FastMCP framework specifically requires Pydantic v2, utilizing `ConfigDict` (not the deprecated `Config` class) and `field_validator` (not the deprecated `validator` decorator) for model configuration. Using v1 constructs will result in import errors or undefined behavior during schema generation in [`python_mcp_server.md`](https://github.com/ComposioHQ/awesome-codex-skills/blob/main/python_mcp_server.md) implementations.

### What happens if a client sends extra fields not defined in the Pydantic model?

If your model uses `ConfigDict(extra="forbid")`, FastMCP rejects the request with a validation error before the tool executes, preventing parameter hallucination. Without this setting, Pydantic ignores extra fields by default, which may lead to silent data loss or unexpected behavior in downstream APIs.

### Where is the JSON Schema for my tool defined?

FastMCP automatically generates the JSON Schema from your Pydantic model when you apply the `@mcp.tool` decorator. The schema is exposed to clients via the standard MCP `tools/list` endpoint, derived from the model's field types, `Field` descriptions, and constraints as implemented in the ComposioHQ/awesome-codex-skills reference architecture.