Tools and Resources in Python: Decorators, Pydantic, and Differences from TypeScript

Capsule description

In the previous capsule you created your first Python MCP server and saw the magic of the decorators: @mcp.tool() extracts the name, description, and schema from your function automatically. Now it's time to go deeper.

This capsule covers the complete implementation of tools and resources in Python: advanced validation with Pydantic, handling complex types, design patterns for robust tools, dynamic resources with URI templates, and the idiomatic differences that make the Python SDK feel different from the TypeScript one.

The goal is that by the end of this capsule, you can implement any tool or resource you need for your Python MCP server, with solid validation and proper error handling.

Advanced tools with @mcp.tool()

Beyond "Hello World"

The tools from the setup were simple: a function, a returned string. Real tools need:

• Validation of complex inputs
• Multiple parameters with varied types
• Robust error handling
• Return of structured content
• Interaction with external services

Tool with multiple parameter types

from mcp.server.fastmcp import FastMCP
import json

mcp = FastMCP("advanced-tools")


@mcp.tool()
async def search_items(
    query: str,
    category: str = "all",
    max_results: int = 10,
    include_metadata: bool = False,
    tags: list[str] | None = None,
) -> str:
    """Searches items in the catalog.

    Allows filtering by category, limiting results,
    including metadata, and filtering by tags.
    """
    results = []
    for i in range(min(max_results, 5)):
        item = {
            "id": i + 1,
            "name": f"Item matching '{query}' #{i + 1}",
            "category": category,
        }
        if include_metadata:
            item["metadata"] = {"relevance": 0.95 - (i * 0.1), "source": "catalog"}
        if tags:
            item["matched_tags"] = tags
        results.append(item)

    return json.dumps(results, indent=2, ensure_ascii=False)

The automatically generated JSON schema:

{
  "type": "object",
  "properties": {
    "query": { "type": "string" },
    "category": { "type": "string", "default": "all" },
    "max_results": { "type": "integer", "default": 10 },
    "include_metadata": { "type": "boolean", "default": false },
    "tags": {
      "anyOf": [
        { "type": "array", "items": { "type": "string" } },
        { "type": "null" }
      ],
      "default": null
    }
  },
  "required": ["query"]
}

The SDK converts Python's type hints to JSON Schema automatically. list[str] | None is converted into a schema with anyOf.

Tool with Pydantic validation

For complex inputs, use Pydantic models as a parameter:

from pydantic import BaseModel, Field


class CreateTaskInput(BaseModel):
    title: str = Field(min_length=1, max_length=200, description="Task title")
    description: str = Field(default="", max_length=2000, description="Detailed description")
    priority: int = Field(default=3, ge=1, le=5, description="Priority from 1 (highest) to 5 (lowest)")
    assignee: str | None = Field(default=None, description="Assigned person")
    tags: list[str] = Field(default_factory=list, description="Tags to organize")


tasks: list[dict] = []


@mcp.tool()
async def create_task(input: CreateTaskInput) -> str:
    """Creates a new task with complete validation."""
    task = {
        "id": len(tasks) + 1,
        "title": input.title,
        "description": input.description,
        "priority": input.priority,
        "assignee": input.assignee,
        "tags": input.tags,
        "status": "pending",
    }
    tasks.append(task)
    return json.dumps({"message": "Task created", "task": task}, indent=2, ensure_ascii=False)

Why Pydantic instead of simple type hints?

Simple type hints validate types but not values. Pydantic adds:

Comparison: Pydantic vs Zod

Zod (TypeScript):

const CreateTaskSchema = z.object({
  title: z.string().min(1).max(200).describe("Task title"),
  description: z.string().max(2000).default("").describe("Detailed description"),
  priority: z.number().int().min(1).max(5).default(3).describe("Priority 1-5"),
  assignee: z.string().optional().describe("Assigned person"),
  tags: z.array(z.string()).default([]).describe("Tags"),
});

server.tool("create_task", "Creates a new task", CreateTaskSchema.shape, async (input) => {
  // ...
});

Pydantic (Python):

class CreateTaskInput(BaseModel):
    title: str = Field(min_length=1, max_length=200, description="Task title")
    description: str = Field(default="", max_length=2000, description="Detailed description")
    priority: int = Field(default=3, ge=1, le=5, description="Priority 1-5")
    assignee: str | None = Field(default=None, description="Assigned person")
    tags: list[str] = Field(default_factory=list, description="Tags")

@mcp.tool()
async def create_task(input: CreateTaskInput) -> str:
    """Creates a new task."""
    # ...

The key difference: in Zod you define the schema separately from the function. In Pydantic, you define a model that is both the schema and the data structure. The Pydantic model is a Python class you can reuse throughout your application — not just for MCP.

Tool with error handling

Tools fail. APIs go down, files don't exist, inputs are invalid. Your tool must handle that:

import httpx


@mcp.tool()
async def fetch_url(url: str, timeout: int = 10) -> str:
    """Fetches the content of a URL.

    Returns the text content of the HTTP response.
    """
    try:
        async with httpx.AsyncClient() as client:
            response = await client.get(url, timeout=timeout)
            response.raise_for_status()
            content_type = response.headers.get("content-type", "")
            if "text" in content_type or "json" in content_type:
                return response.text[:5000]
            return f"Response received ({response.status_code}), but the content isn't text: {content_type}"
    except httpx.TimeoutException:
        return f"Error: timeout after {timeout} seconds connecting to {url}"
    except httpx.HTTPStatusError as e:
        return f"HTTP error {e.response.status_code}: {e.response.text[:500]}"
    except httpx.RequestError as e:
        return f"Connection error: {e}"
    except Exception as e:
        return f"Unexpected error: {type(e).__name__}: {e}"

Recommended pattern: Don't raise exceptions from tools. Return error messages as text. The model can read the error and try to correct its approach. If you raise an exception, the SDK catches it but the message can be less useful.

Tool with the dry-run pattern

import os


@mcp.tool()
async def delete_files(
    directory: str,
    pattern: str,
    dry_run: bool = True,
) -> str:
    """Deletes files that match a pattern.

    By default runs in dry_run mode (only shows what would be deleted).
    Use dry_run=false to execute the actual deletion.
    """
    import fnmatch

    if not os.path.isdir(directory):
        return f"Error: '{directory}' is not a valid directory."

    matches = []
    for filename in os.listdir(directory):
        if fnmatch.fnmatch(filename, pattern):
            matches.append(filename)

    if not matches:
        return f"No files found matching '{pattern}' in {directory}."

    if dry_run:
        file_list = "\n".join(f"  - {f}" for f in matches)
        return f"[DRY RUN] {len(matches)} files would be deleted:\n{file_list}\n\nUse dry_run=false to execute."

    deleted = []
    errors = []
    for filename in matches:
        try:
            filepath = os.path.join(directory, filename)
            os.remove(filepath)
            deleted.append(filename)
        except OSError as e:
            errors.append(f"{filename}: {e}")

    result = f"Deleted: {len(deleted)} files."
    if errors:
        result += f"\nErrors: {len(errors)}\n" + "\n".join(f"  - {e}" for e in errors)
    return result

The dry-run pattern is fundamental for destructive tools. The model first runs with dry_run=True, the user sees the preview, and then can confirm with dry_run=False.

Advanced resources with @mcp.resource()

Static vs dynamic resources

Static resource — the URI is fixed:

@mcp.resource("config://app/version")
async def app_version() -> str:
    """Current version of the application."""
    return "2.5.1"

Dynamic resource with a template — the URI has parameters:

@mcp.resource("users://{user_id}/profile")
async def user_profile(user_id: str) -> str:
    """Profile of a specific user."""
    users_db = {
        "1": {"name": "Ann", "role": "admin", "email": "ann@example.com"},
        "2": {"name": "Carl", "role": "developer", "email": "carl@example.com"},
    }
    user = users_db.get(user_id)
    if not user:
        return json.dumps({"error": f"User {user_id} not found"})
    return json.dumps(user, indent=2)

The {user_id} in the URI is a template. When a client requests users://42/profile, the SDK extracts user_id="42" and passes it to your function.

Resource that returns structured JSON

@mcp.resource("metrics://server/health")
async def server_health() -> str:
    """Server health metrics."""
    import psutil

    health = {
        "status": "healthy",
        "cpu_percent": psutil.cpu_percent(),
        "memory": {
            "total_gb": round(psutil.virtual_memory().total / (1024**3), 2),
            "used_percent": psutil.virtual_memory().percent,
        },
        "disk": {
            "total_gb": round(psutil.disk_usage("/").total / (1024**3), 2),
            "used_percent": psutil.disk_usage("/").percent,
        },
    }
    return json.dumps(health, indent=2)

Comparison: Resources in Python vs TypeScript

TypeScript:

server.resource(
  "user-profile",
  "users://{user_id}/profile",
  { description: "Profile of a user", mimeType: "application/json" },
  async (uri) => {
    const userId = uri.pathname.split("/")[1];
    const user = await getUserById(userId);
    return {
      contents: [{
        uri: uri.href,
        mimeType: "application/json",
        text: JSON.stringify(user),
      }],
    };
  }
);

Python:

@mcp.resource("users://{user_id}/profile")
async def user_profile(user_id: str) -> str:
    """Profile of a user."""
    user = await get_user_by_id(user_id)
    return json.dumps(user, indent=2)

Key differences:

• Python extracts the URI template's parameters automatically as function arguments
• You don't need to parse the URI manually
• The return is a simple string — the SDK wraps it in the MCP format
• The description comes from the docstring

Design patterns for tools

Pattern 1: Complete CRUD

from pydantic import BaseModel, Field
from datetime import datetime

mcp = FastMCP("todo-crud")

todos: list[dict] = []
next_id = 1


class TodoCreate(BaseModel):
    title: str = Field(min_length=1, max_length=200)
    description: str = Field(default="")
    priority: int = Field(default=3, ge=1, le=5)


class TodoUpdate(BaseModel):
    title: str | None = Field(default=None, min_length=1, max_length=200)
    description: str | None = None
    priority: int | None = Field(default=None, ge=1, le=5)
    completed: bool | None = None


@mcp.tool()
async def create_todo(input: TodoCreate) -> str:
    """Creates a new TODO."""
    global next_id
    todo = {
        "id": next_id,
        "title": input.title,
        "description": input.description,
        "priority": input.priority,
        "completed": False,
        "created_at": datetime.now().isoformat(),
    }
    next_id += 1
    todos.append(todo)
    return json.dumps({"message": "TODO created", "todo": todo}, indent=2)


@mcp.tool()
async def get_todo(todo_id: int) -> str:
    """Gets a TODO by its ID."""
    for todo in todos:
        if todo["id"] == todo_id:
            return json.dumps(todo, indent=2)
    return json.dumps({"error": f"TODO with ID {todo_id} not found"})


@mcp.tool()
async def update_todo(todo_id: int, updates: TodoUpdate) -> str:
    """Updates an existing TODO."""
    for todo in todos:
        if todo["id"] == todo_id:
            update_data = updates.model_dump(exclude_none=True)
            todo.update(update_data)
            todo["updated_at"] = datetime.now().isoformat()
            return json.dumps({"message": "TODO updated", "todo": todo}, indent=2)
    return json.dumps({"error": f"TODO with ID {todo_id} not found"})


@mcp.tool()
async def delete_todo(todo_id: int) -> str:
    """Deletes a TODO by its ID."""
    global todos
    original_len = len(todos)
    todos = [t for t in todos if t["id"] != todo_id]
    if len(todos) < original_len:
        return f"TODO with ID {todo_id} deleted."
    return f"Error: TODO with ID {todo_id} not found."


@mcp.tool()
async def list_todos(
    status: str = "all",
    sort_by: str = "created_at",
) -> str:
    """Lists TODOs with optional filters.

    status: 'all', 'completed', or 'pending'.
    sort_by: 'created_at', 'priority', or 'title'.
    """
    filtered = todos
    if status == "completed":
        filtered = [t for t in todos if t["completed"]]
    elif status == "pending":
        filtered = [t for t in todos if not t["completed"]]

    if sort_by in ("priority", "title", "created_at"):
        filtered = sorted(filtered, key=lambda t: t.get(sort_by, ""))

    return json.dumps({"total": len(filtered), "todos": filtered}, indent=2)

Pattern 2: Tool that wraps an external API

import httpx


@mcp.tool()
async def get_weather(city: str, units: str = "metric") -> str:
    """Gets the current weather for a city.

    units: 'metric' (Celsius) or 'imperial' (Fahrenheit).
    """
    api_key = os.environ.get("OPENWEATHER_API_KEY")
    if not api_key:
        return "Error: OPENWEATHER_API_KEY not configured. Set the environment variable."

    try:
        async with httpx.AsyncClient() as client:
            response = await client.get(
                "https://api.openweathermap.org/data/2.5/weather",
                params={"q": city, "appid": api_key, "units": units, "lang": "en"},
                timeout=10,
            )
            response.raise_for_status()
            data = response.json()

        temp = data["main"]["temp"]
        feels_like = data["main"]["feels_like"]
        description = data["weather"][0]["description"]
        humidity = data["main"]["humidity"]
        unit_symbol = "°C" if units == "metric" else "°F"

        return (
            f"Weather in {city}:\n"
            f"  Temperature: {temp}{unit_symbol} (feels like: {feels_like}{unit_symbol})\n"
            f"  Condition: {description}\n"
            f"  Humidity: {humidity}%"
        )
    except httpx.HTTPStatusError as e:
        if e.response.status_code == 404:
            return f"Error: city '{city}' not found."
        return f"API error: {e.response.status_code}"
    except httpx.RequestError as e:
        return f"Connection error: {e}"

Pattern 3: Complementary Tool + Resource

import json

project_files: dict[str, str] = {}


@mcp.resource("project://files/list")
async def list_project_files() -> str:
    """List of files in the project."""
    return json.dumps(list(project_files.keys()), indent=2)


@mcp.resource("project://files/{filename}")
async def read_project_file(filename: str) -> str:
    """Content of a project file."""
    content = project_files.get(filename)
    if content is None:
        return f"Error: file '{filename}' not found."
    return content


@mcp.tool()
async def write_project_file(filename: str, content: str) -> str:
    """Writes a file to the project."""
    project_files[filename] = content
    return f"File '{filename}' written ({len(content)} characters)."


@mcp.tool()
async def analyze_project() -> str:
    """Analyzes all the project's files.

    Returns statistics for each file.
    """
    if not project_files:
        return "No files in the project."

    stats = []
    for name, content in project_files.items():
        lines = content.count("\n") + 1
        words = len(content.split())
        stats.append({"file": name, "lines": lines, "words": words, "chars": len(content)})

    return json.dumps({"total_files": len(stats), "files": stats}, indent=2)

The project://files/list resource lets the model see what files exist. The project://files/{filename} resource lets it read a specific file. The write_project_file tool lets it create files. And analyze_project acts on the data. Resource to read, tool to act.

Decorators vs Classes: the fundamental difference

The TypeScript approach: explicit and verbose

server.tool(
  "analyze_text",                    // name: explicit
  "Analyzes a text and returns statistics",  // description: explicit
  {                                   // schema: explicit with Zod
    text: z.string().min(1),
    include_sentiment: z.boolean().default(false),
  },
  async ({ text, include_sentiment }) => {  // handler: separate
    const words = text.split(/\s+/).length;
    const sentences = text.split(/[.!?]+/).filter(Boolean).length;

    let result = `Words: ${words}, Sentences: ${sentences}`;
    if (include_sentiment) {
      result += `, Sentiment: neutral`;
    }

    return { content: [{ type: "text", text: result }] };
  }
);

Four separate pieces: name, description, schema, handler.

The Python approach: inferred and concise

@mcp.tool()
async def analyze_text(text: str, include_sentiment: bool = False) -> str:
    """Analyzes a text and returns statistics."""
    words = len(text.split())
    sentences = len([s for s in text.split(".") if s.strip()])

    result = f"Words: {words}, Sentences: {sentences}"
    if include_sentiment:
        result += ", Sentiment: neutral"
    return result

A single piece: the function. Everything else is inferred.

Neither is "better" — they're different philosophies. TypeScript is explicit: you see exactly what's registered. Python is conventional: if you follow the conventions (docstrings, type hints), the framework does the work.

Exercises

Exercise 1: Tool with a nested Pydantic model (Medium)

Create a create_event tool that takes a Pydantic model with:

• title (str, required, 1-100 chars)
• date (str, required, ISO format)
• location with name (str) and address (str, optional)
• attendees (list of strings, maximum 50)

from pydantic import BaseModel, Field
from mcp.server.fastmcp import FastMCP
import json

mcp = FastMCP("event-manager")

events: list[dict] = []


class Location(BaseModel):
    name: str = Field(description="Name of the place")
    address: str | None = Field(default=None, description="Full address")


class CreateEventInput(BaseModel):
    title: str = Field(min_length=1, max_length=100, description="Event title")
    date: str = Field(description="Date in ISO format (YYYY-MM-DD)")
    location: Location = Field(description="Event location")
    attendees: list[str] = Field(
        default_factory=list, max_length=50, description="List of attendees"
    )


@mcp.tool()
async def create_event(input: CreateEventInput) -> str:
    """Creates a new event with location and attendees."""
    event = {
        "id": len(events) + 1,
        "title": input.title,
        "date": input.date,
        "location": input.location.model_dump(),
        "attendees": input.attendees,
        "attendee_count": len(input.attendees),
    }
    events.append(event)
    return json.dumps({"message": "Event created", "event": event}, indent=2, ensure_ascii=False)


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

Exercise 2: Resource with a dynamic URI template (Medium)

Create an MCP server with:

• A resource inventory://products/{category} that returns products filtered by category
• A resource inventory://products/{category}/{product_id} that returns a specific product
• A tool add_product to add products to the inventory

from mcp.server.fastmcp import FastMCP
import json

mcp = FastMCP("inventory")

products: list[dict] = [
    {"id": 1, "name": "Laptop Pro", "category": "electronics", "price": 1299.99},
    {"id": 2, "name": "Mechanical Keyboard", "category": "electronics", "price": 89.99},
    {"id": 3, "name": "Python Cookbook", "category": "books", "price": 45.00},
    {"id": 4, "name": "Standing Desk", "category": "furniture", "price": 599.99},
]


@mcp.resource("inventory://products/{category}")
async def products_by_category(category: str) -> str:
    """Products filtered by category."""
    filtered = [p for p in products if p["category"] == category]
    return json.dumps(
        {"category": category, "count": len(filtered), "products": filtered},
        indent=2,
    )


@mcp.resource("inventory://products/{category}/{product_id}")
async def product_detail(category: str, product_id: str) -> str:
    """Detail of a specific product."""
    for p in products:
        if p["category"] == category and str(p["id"]) == product_id:
            return json.dumps(p, indent=2)
    return json.dumps({"error": f"Product {product_id} not found in '{category}'"})


@mcp.tool()
async def add_product(name: str, category: str, price: float) -> str:
    """Adds a product to the inventory."""
    product = {
        "id": max((p["id"] for p in products), default=0) + 1,
        "name": name,
        "category": category,
        "price": price,
    }
    products.append(product)
    return json.dumps({"message": "Product added", "product": product}, indent=2)


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

Exercise 3: Convert tools from TypeScript to Python (Medium)

Convert the following MCP server from TypeScript to idiomatic Python:

const server = new McpServer({ name: "file-stats", version: "1.0.0" });

server.tool("count_lines", "Counts lines of a file", {
  filepath: z.string().describe("Path to the file"),
  skip_empty: z.boolean().default(false).describe("Ignore empty lines"),
}, async ({ filepath, skip_empty }) => {
  const content = await fs.readFile(filepath, "utf-8");
  let lines = content.split("\n");
  if (skip_empty) {
    lines = lines.filter(line => line.trim().length > 0);
  }
  return {
    content: [{
      type: "text",
      text: JSON.stringify({ filepath, total_lines: lines.length, skip_empty }),
    }],
  };
});

server.tool("file_info", "Information about a file", {
  filepath: z.string(),
}, async ({ filepath }) => {
  const stats = await fs.stat(filepath);
  return {
    content: [{
      type: "text",
      text: JSON.stringify({
        size_bytes: stats.size,
        created: stats.birthtime.toISOString(),
        modified: stats.mtime.toISOString(),
        is_directory: stats.isDirectory(),
      }),
    }],
  };
});

import os
import json
from datetime import datetime
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("file-stats")


@mcp.tool()
async def count_lines(filepath: str, skip_empty: bool = False) -> str:
    """Counts lines of a file.

    Optionally ignores empty lines.
    """
    try:
        with open(filepath, "r", encoding="utf-8") as f:
            lines = f.readlines()

        if skip_empty:
            lines = [line for line in lines if line.strip()]

        return json.dumps(
            {"filepath": filepath, "total_lines": len(lines), "skip_empty": skip_empty},
            indent=2,
        )
    except FileNotFoundError:
        return json.dumps({"error": f"File not found: {filepath}"})
    except PermissionError:
        return json.dumps({"error": f"No permissions to read: {filepath}"})


@mcp.tool()
async def file_info(filepath: str) -> str:
    """Information about a file: size, dates, type."""
    try:
        stats = os.stat(filepath)
        return json.dumps(
            {
                "size_bytes": stats.st_size,
                "created": datetime.fromtimestamp(stats.st_ctime).isoformat(),
                "modified": datetime.fromtimestamp(stats.st_mtime).isoformat(),
                "is_directory": os.path.isdir(filepath),
            },
            indent=2,
        )
    except FileNotFoundError:
        return json.dumps({"error": f"File not found: {filepath}"})


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

Exercise 4: Tool with an enum and strict validation (Hard)

Create a process_data tool that:

• Takes a DataProcessInput with a format field that's an Enum (csv, json, xml)
• Takes data as a string
• Takes options as a Pydantic model with skip_header (bool), delimiter (str), and encoding (str)
• Validates that delimiter is only used when format is csv

from enum import Enum
from pydantic import BaseModel, Field, model_validator
from mcp.server.fastmcp import FastMCP
import json

mcp = FastMCP("data-processor")


class DataFormat(str, Enum):
    CSV = "csv"
    JSON = "json"
    XML = "xml"


class ProcessOptions(BaseModel):
    skip_header: bool = Field(default=False, description="Skip the first line (CSV only)")
    delimiter: str = Field(default=",", description="Delimiter (CSV only)")
    encoding: str = Field(default="utf-8", description="Text encoding")


class DataProcessInput(BaseModel):
    format: DataFormat = Field(description="Format of the data")
    data: str = Field(min_length=1, description="Data to process")
    options: ProcessOptions = Field(default_factory=ProcessOptions)

    @model_validator(mode="after")
    def validate_options(self):
        if self.format != DataFormat.CSV and self.options.delimiter != ",":
            raise ValueError("The delimiter only applies to the CSV format")
        if self.format != DataFormat.CSV and self.options.skip_header:
            raise ValueError("skip_header only applies to the CSV format")
        return self


@mcp.tool()
async def process_data(input: DataProcessInput) -> str:
    """Processes data in CSV, JSON, or XML format."""
    if input.format == DataFormat.CSV:
        lines = input.data.strip().split("\n")
        if input.options.skip_header and len(lines) > 1:
            lines = lines[1:]
        rows = [line.split(input.options.delimiter) for line in lines]
        return json.dumps(
            {"format": "csv", "rows": len(rows), "columns": len(rows[0]) if rows else 0, "data": rows},
            indent=2,
        )

    elif input.format == DataFormat.JSON:
        try:
            parsed = json.loads(input.data)
            item_count = len(parsed) if isinstance(parsed, list) else 1
            return json.dumps({"format": "json", "items": item_count, "valid": True}, indent=2)
        except json.JSONDecodeError as e:
            return json.dumps({"format": "json", "valid": False, "error": str(e)})

    elif input.format == DataFormat.XML:
        tag_count = input.data.count("<") // 2
        return json.dumps({"format": "xml", "estimated_tags": tag_count}, indent=2)

    return json.dumps({"error": "Unsupported format"})


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

Exercise 5: Complete server with the 3 primitives (Hard)

Create an MCP server "bookmark-manager" with:

• Resources: bookmarks://all, bookmarks://category/{cat}
• Tools: add_bookmark, delete_bookmark, search_bookmarks
• Prompt: organize_bookmarks that asks the model to organize the bookmarks by category

from mcp.server.fastmcp import FastMCP
from pydantic import BaseModel, Field
from datetime import datetime
import json

mcp = FastMCP("bookmark-manager")

bookmarks: list[dict] = []
next_id = 1


class BookmarkInput(BaseModel):
    url: str = Field(description="Bookmark URL")
    title: str = Field(min_length=1, max_length=200, description="Descriptive title")
    category: str = Field(default="uncategorized", description="Category")
    tags: list[str] = Field(default_factory=list, description="Tags for searching")


@mcp.resource("bookmarks://all")
async def all_bookmarks() -> str:
    """All the stored bookmarks."""
    return json.dumps(
        {"total": len(bookmarks), "bookmarks": bookmarks}, indent=2, ensure_ascii=False
    )


@mcp.resource("bookmarks://category/{category}")
async def bookmarks_by_category(category: str) -> str:
    """Bookmarks filtered by category."""
    filtered = [b for b in bookmarks if b["category"] == category]
    return json.dumps(
        {"category": category, "count": len(filtered), "bookmarks": filtered},
        indent=2,
        ensure_ascii=False,
    )


@mcp.tool()
async def add_bookmark(input: BookmarkInput) -> str:
    """Adds a new bookmark."""
    global next_id
    bookmark = {
        "id": next_id,
        "url": input.url,
        "title": input.title,
        "category": input.category,
        "tags": input.tags,
        "created_at": datetime.now().isoformat(),
    }
    next_id += 1
    bookmarks.append(bookmark)
    return json.dumps({"message": "Bookmark added", "bookmark": bookmark}, indent=2, ensure_ascii=False)


@mcp.tool()
async def delete_bookmark(bookmark_id: int) -> str:
    """Deletes a bookmark by ID."""
    global bookmarks
    before = len(bookmarks)
    bookmarks = [b for b in bookmarks if b["id"] != bookmark_id]
    if len(bookmarks) < before:
        return f"Bookmark {bookmark_id} deleted."
    return f"Error: bookmark {bookmark_id} not found."


@mcp.tool()
async def search_bookmarks(query: str) -> str:
    """Searches bookmarks by title, URL, or tags."""
    query_lower = query.lower()
    results = [
        b
        for b in bookmarks
        if query_lower in b["title"].lower()
        or query_lower in b["url"].lower()
        or any(query_lower in tag.lower() for tag in b["tags"])
    ]
    return json.dumps({"query": query, "results": len(results), "bookmarks": results}, indent=2, ensure_ascii=False)


@mcp.prompt()
async def organize_bookmarks() -> str:
    """Asks the model to organize the bookmarks by category."""
    data = json.dumps(bookmarks, indent=2, ensure_ascii=False) if bookmarks else "No bookmarks."
    return f"""Analyze the following bookmarks and suggest a reorganization by categories:

{data}

Please:
1. Group the bookmarks into logical categories
2. Suggest new categories if the current ones aren't descriptive
3. Identify duplicate or similar bookmarks
4. Recommend useful tags for each bookmark
5. Use the delete_bookmark and add_bookmark tools to implement the changes"""


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

Troubleshooting

"The tool doesn't receive the correct parameters"

Cause: The model sends parameters that don't match your function's type hints.

Solution:

@mcp.tool()
async def my_tool(
    name: str,           # ✅ Explicit type
    count: int = 5,      # ✅ Explicit default
    tags: list[str] | None = None,  # ✅ Optional with default
) -> str:
    """Description that tells the model exactly which parameters to use."""
    ...

"Pydantic ValidationError when invoking the tool"

Cause: The sent data doesn't pass Pydantic's validation.

Solution:

class MyInput(BaseModel):
    value: int = Field(ge=0, le=100)

@mcp.tool()
async def my_tool(input: MyInput) -> str:
    """The valid range is 0-100. Out-of-range values will be rejected."""
    return str(input.value)

"The resource template doesn't extract the URI parameters"

Cause: The parameter names in the URI template don't match the ones in the function.

Solution:

@mcp.resource("data://{item_type}/{item_id}")
async def get_item(item_type: str, item_id: str) -> str:
    ...

"TypeError: object str can't be used in 'await' expression"

Cause: You're using await on a function that isn't async, or you're calling a sync function with await.

Solution:

@mcp.tool()
async def my_tool(path: str) -> str:
    content = open(path).read()  # sync, no await
    return content

Summary

In this capsule you learned:

• Advanced tools with multiple parameters, complex types, and Pydantic models
• Pydantic vs Zod: both validate, but Pydantic integrates with Python's type hints
• Dynamic resources with URI templates that extract parameters automatically
• Design patterns: complete CRUD, API wrapper, dry-run, complementary tool + resource
• Decorators vs Classes: Python infers the name, description, and schema; TypeScript declares them explicitly
• Error handling as returned text (not exceptions) so the model can interpret errors

Next capsule: Async patterns in MCP Python — asyncio, async context managers, generators, and why async matters when your MCP server connects with APIs and databases.

Additional resources

1. Pydantic Field Validators — Advanced validation with Pydantic
2. MCP Python SDK — Tools — Official tools implementation
3. Python Type Hints Cheat Sheet — Quick reference for type hints
4. Zod Documentation — To compare with Pydantic
5. MCP Specification — Resources — Official resources spec
6. MCP Specification — Tools — Official tools spec

Next capsule: Async Patterns in MCP — asyncio in the context of MCP servers, connections to APIs, and async error handling.