Skip to main content

Master MCP Python Tools

Calculating read time…

Imagine you have an AI assistant that can do anything — check your calendar, search databases, send emails, or even control smart home devices.

But here's the challenge: how do you connect all these capabilities to your AI in a way that's clean, maintainable, and actually works?

That's where the Model Context Protocol (MCP) comes in, and specifically — how to expose your Python functions as MCP tools.

In this comprehensive guide, we'll cover everything you need to know:

  • How to expose Python functions as MCP tools
  • Schema validation techniques
  • Returning structured results
  • Handling multiple clients simultaneously

By the end, you'll be able to build production-ready MCP servers that power intelligent AI applications.

💡 What You'll Learn

This is a hands-on tutorial. We'll build real code, not just talk about concepts.

You'll learn by doing, with complete working examples you can run immediately.

🎯 Part 1: Exposing Python Functions as MCP Tools

Let's start with the fundamental question: What does it mean to "expose a function as an MCP tool"?

The Simple Analogy 🔧

Think of your Python functions as physical tools in a toolbox — a hammer, screwdriver, drill, etc.

But your AI assistant can't just reach into the toolbox and grab a tool. It needs:

  • A catalog listing what tools are available
  • Instructions on how to use each tool
  • A standardized way to request a tool and get the result

MCP provides exactly this infrastructure. It turns your Python functions into "tools" that AI can discover and use.

Your First MCP Tool: A Simple Calculator

Let's build our first MCP tool from scratch. We'll create a calculator function and expose it via MCP.

Step 1: Install the MCP Python SDK 

pip install mcp

Step 2: Create your first MCP server 

from mcp.server.fastmcp import FastMCP

# Create an MCP server instance
mcp = FastMCP("Calculator Server")

# Expose a Python function as an MCP tool
@mcp.tool()
def add_numbers(a: int, b: int) -> int:
    """Add two numbers together and return the result."""
    return a + b

@mcp.tool()
def multiply_numbers(a: float, b: float) -> float:
    """Multiply two numbers together."""
    return a * b

# Run the server
if __name__ == "__main__":
    mcp.run(transport="stdio")

That's it! You just created your first MCP server with two tools.

Notice what happened here:

  • The @mcp.tool() decorator converts a regular Python function into an MCP tool
  • Type hints (a: int, b: int) tell MCP what parameter types to expect
  • The docstring becomes the tool's description that the AI reads
  • The return type (-> int) specifies what the tool returns

✅ Best Practice: Write Clear Docstrings

The docstring is crucial! It's how the AI decides whether to use your tool.

Instead of "Add numbers" write "Add two numbers together and return the sum. Use this when the user asks for addition, total, or sum calculations."

Understanding the Magic Behind @mcp.tool()

When you use @mcp.tool(), FastMCP automatically:

  1. Inspects your function signature using Python's inspect module
  2. Extracts parameter names, types, and default values
  3. Generates a JSON Schema describing valid inputs
  4. Creates validation logic using Pydantic
  5. Registers the tool in the MCP server's catalog

All of this happens automatically. You just write normal Python code!

Example: Weather Lookup Tool

Let's create a more realistic example — a weather lookup tool. 

import httpx
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("Weather Service")

@mcp.tool()
async def get_weather(
    city: str,
    country: str = "US",
    units: str = "metric"
) -> dict:
    """
    Get current weather for a specific city.
    
    Returns temperature, conditions, humidity, and wind speed.
    Use this when users ask about current weather conditions.
    """
    # Call a weather API (example using a fictional endpoint)
    async with httpx.AsyncClient() as client:
        response = await client.get(
            f"https://api.weather.example.com/current",
            params={"city": city, "country": country, "units": units}
        )
        return response.json()

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

Notice the improvements here:

  • We used async def for asynchronous operations (API calls)
  • Default parameters (country: str = "US") make some inputs optional
  • Return type is dict for structured data
  • The docstring explains exactly when to use this tool

Tools With Complex Parameters

What if your tool needs complex input structures? Use Pydantic models! 

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

mcp = FastMCP("Email Server")

class EmailMessage(BaseModel):
    """Structure for an email message."""
    recipient: str = Field(description="Email address of the recipient")
    subject: str = Field(description="Email subject line")
    body: str = Field(description="Email message body")
    cc: list[str] = Field(default=[], description="CC recipients")
    attachments: list[str] = Field(default=[], description="File paths to attach")

@mcp.tool()
async def send_email(message: EmailMessage) -> dict:
    """
    Send an email message with optional CC and attachments.
    
    Use this when the user wants to send an email, compose a message,
    or contact someone via email.
    """
    # Email sending logic here
    print(f"Sending email to {message.recipient}")
    print(f"Subject: {message.subject}")
    print(f"Body: {message.body}")
    
    return {
        "status": "sent",
        "message_id": "abc123",
        "recipient": message.recipient
    }

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

This approach gives you:

  • Automatic validation of complex nested structures
  • Clear field descriptions for the AI
  • Type safety throughout your code
  • Easy-to-maintain schemas

❌ Common Mistake: Missing Type Hints

Without type hints, MCP cannot generate proper schemas.

Bad: def add(a, b):
Good: def add(a: int, b: int) -> int:

🔒 Part 2: Schema Validation — Keeping Data Safe

Schema validation is your first line of defense against bad input.

It ensures that tools receive the right type of data, in the right format, preventing crashes and security issues.

Why Validation Matters

Imagine an AI calls your "delete_file" tool with a file path like ../../etc/passwd.

Without validation, this could be a security disaster!

Proper schema validation prevents:

  • Type errors (passing string instead of integer)
  • Missing required fields
  • Invalid values (negative age, future birth date)
  • Security vulnerabilities (path traversal, SQL injection)
  • Malformed data structures

Built-In Validation With Type Hints

FastMCP automatically validates based on Python type hints: 

@mcp.tool()
def process_age(age: int) -> str:
    """Process user age."""
    return f"You are {age} years old"

# Valid call: process_age(25) ✅
# Invalid call: process_age("twenty-five") ❌ — Validation error!
# Invalid call: process_age(25.5) ❌ — Must be integer, not float!

The MCP server automatically rejects invalid calls before your code runs.

Advanced Validation With Pydantic

For complex validation rules, use Pydantic's powerful features: 

from pydantic import BaseModel, Field, field_validator, EmailStr
from datetime import date
from typing import Literal

class UserRegistration(BaseModel):
    """User registration data with comprehensive validation."""
    
    email: EmailStr = Field(description="User's email address")
    age: int = Field(ge=18, le=120, description="Age must be 18-120")
    username: str = Field(
        min_length=3,
        max_length=20,
        pattern="^[a-zA-Z0-9_]+$",
        description="Username (alphanumeric and underscore only)"
    )
    country: Literal["US", "UK", "CA", "AU"] = Field(
        description="Country code"
    )
    birth_date: date = Field(description="Birth date")
    
    @field_validator('birth_date')
    @classmethod
    def validate_birth_date(cls, v):
        """Ensure birth date is in the past."""
        if v >= date.today():
            raise ValueError("Birth date must be in the past")
        return v
    
    @field_validator('username')
    @classmethod
    def no_profanity(cls, v):
        """Check for inappropriate usernames."""
        forbidden = ['admin', 'root', 'moderator']
        if v.lower() in forbidden:
            raise ValueError(f"Username '{v}' is reserved")
        return v

@mcp.tool()
def register_user(data: UserRegistration) -> dict:
    """
    Register a new user with validated data.
    
    All fields are automatically validated before registration.
    """
    return {
        "status": "success",
        "user_id": "user_12345",
        "email": data.email
    }

This validation schema ensures:

  • email is a valid email format (using EmailStr)
  • age is between 18-120 (using ge/le constraints)
  • username is 3-20 chars, alphanumeric only (regex pattern)
  • country is one of the allowed values (Literal type)
  • birth_date is in the past (custom validator)
  • No reserved usernames allowed (custom validator)

✅ Validation Best Practices

1. Validate at the boundary: Check inputs before processing
2. Fail fast: Reject invalid data immediately
3. Clear error messages: Help users fix problems
4. Whitelist, don't blacklist: Allow only known-good patterns

Real-World Example: File Operations

File operations need careful validation to prevent security issues: 

from pathlib import Path
from pydantic import BaseModel, field_validator

class FileOperation(BaseModel):
    """Validated file operation parameters."""
    
    file_path: str = Field(description="Path to the file")
    operation: Literal["read", "write", "delete"] = Field(
        description="Operation to perform"
    )
    
    @field_validator('file_path')
    @classmethod
    def validate_safe_path(cls, v):
        """Ensure path is safe and within allowed directory."""
        path = Path(v).resolve()
        allowed_dir = Path("/safe/data/directory").resolve()
        
        # Prevent path traversal attacks
        if not str(path).startswith(str(allowed_dir)):
            raise ValueError("Path must be within allowed directory")
        
        # Prevent accessing hidden files
        if any(part.startswith('.') for part in path.parts):
            raise ValueError("Cannot access hidden files")
        
        return v

@mcp.tool()
def manage_file(operation: FileOperation) -> dict:
    """
    Safely manage files with comprehensive security checks.
    """
    # Your validated file operation here
    return {"status": "success", "path": operation.file_path}

This prevents common vulnerabilities like:

  • Path traversal (../../etc/passwd)
  • Accessing hidden files (.ssh/id_rsa)
  • Operating outside allowed directories

Validation Error Handling

When validation fails, FastMCP automatically returns clear error messages: 

# If AI tries to call with invalid data:
{
  "error": {
    "code": "INVALID_PARAMS",
    "message": "Validation error",
    "data": {
      "age": "value must be between 18 and 120",
      "email": "invalid email format"
    }
  }
}

The AI can read these errors and correct its input automatically.

📦 Part 3: Returning Structured Results

Getting data into your tool is only half the battle. Returning data in a structured, usable format is equally important.

Why Structured Results Matter

Imagine asking an AI: "What's the weather in London?"

Unstructured response: 

"The weather in London is partly cloudy with a temperature 
of 18 degrees Celsius and 65% humidity"

This is fine for humans, but terrible for machines.

Structured response: 

{
  "city": "London",
  "temperature": 18,
  "unit": "celsius",
  "conditions": "partly cloudy",
  "humidity": 65,
  "wind_speed": 12
}

Now the AI (or any application) can:

  • Extract specific values programmatically
  • Compare weather across cities
  • Make decisions based on temperature thresholds
  • Display data in charts or tables

Automatic Structured Output

FastMCP automatically creates structured output when you return certain types: 

from pydantic import BaseModel

class WeatherData(BaseModel):
    """Structured weather information."""
    city: str
    temperature: float
    unit: str
    conditions: str
    humidity: int
    wind_speed: float

@mcp.tool()
async def get_weather(city: str) -> WeatherData:
    """Get current weather as structured data."""
    # Fetch weather from API
    return WeatherData(
        city=city,
        temperature=18.5,
        unit="celsius",
        conditions="partly cloudy",
        humidity=65,
        wind_speed=12.3
    )

FastMCP automatically:

  1. Validates the return value against the schema
  2. Serializes it to JSON
  3. Includes it in the MCP response as structured content
  4. Also provides a text representation for backward compatibility

💡 Key Insight: Dual Format

MCP returns BOTH structured data (for machines) and text content (for humans/backward compatibility).

This means older clients still work, while modern clients get rich structured data.

Complex Structured Results

You can return complex nested structures: 

from pydantic import BaseModel
from typing import List

class Product(BaseModel):
    """Individual product information."""
    id: str
    name: str
    price: float
    in_stock: bool

class SearchResults(BaseModel):
    """Search results with metadata."""
    query: str
    total_results: int
    page: int
    products: List[Product]

@mcp.tool()
def search_products(
    query: str,
    max_results: int = 10
) -> SearchResults:
    """
    Search product catalog and return structured results.
    """
    # Simulate database search
    products = [
        Product(id="p1", name="Laptop", price=999.99, in_stock=True),
        Product(id="p2", name="Mouse", price=29.99, in_stock=True),
        Product(id="p3", name="Keyboard", price=79.99, in_stock=False),
    ]
    
    return SearchResults(
        query=query,
        total_results=len(products),
        page=1,
        products=products[:max_results]
    )

The AI receives a perfectly structured response it can work with programmatically.

Handling Primitive Returns

When returning primitives (int, str, bool), add a return type annotation: 

@mcp.tool()
def calculate_sum(a: int, b: int) -> int:
    """Calculate sum of two numbers."""
    return a + b

# Returns structured content: {"result": 8}

Without the -> int annotation, only text content would be returned.

Advanced: Custom Result Format

For maximum control, return CallToolResult directly: 

from mcp.types import CallToolResult, TextContent, StructuredContent

@mcp.tool()
def advanced_calculation(x: int) -> CallToolResult:
    """Calculation with custom result format."""
    result = x * 2
    
    return CallToolResult(
        content=[
            TextContent(
                type="text",
                text=f"The result is {result}"
            )
        ],
        structuredContent=StructuredContent(
            data={"result": result, "input": x, "operation": "multiply"}
        ),
        _meta={"processing_time_ms": 42, "cached": False}
    )

This gives you control over:

  • Multiple content blocks (text, images, etc.)
  • Structured data format
  • Metadata (not shown to AI, only to client app)

✅ Structured Results Best Practices

1. Use Pydantic models for complex returns
2. Always add return type hints (-> int, -> dict, etc.)
3. Include relevant metadata (timestamps, counts, status)
4. Keep structures flat when possible for easier parsing

🌐 Part 4: Handling Multiple Clients

So far, we've built tools for a single client. But what happens when multiple AI agents or users connect simultaneously?

This is where session management and concurrency handling become critical.

The Challenge of Multiple Clients

Imagine two users using your MCP server at the same time:

  • User A: "Search my calendar for meetings today"
  • User B: "Delete all my calendar events"

Without proper session management, User B might accidentally delete User A's calendar!

We need to:

  1. Isolate each client's session
  2. Maintain separate state for each client
  3. Handle concurrent requests safely
  4. Manage shared resources (database connections, API rate limits)

Session Management Fundamentals

MCP has built-in session management. Each client connection gets a unique session.

Here's how to use it: 

from mcp.server.fastmcp import FastMCP, Context
from mcp.server.session import ServerSession
from typing import Dict

mcp = FastMCP("Multi-Client Server")

# Store session-specific data
user_preferences: Dict[str, dict] = {}

@mcp.tool()
async def set_preference(
    key: str,
    value: str,
    ctx: Context[ServerSession, None]
) -> str:
    """
    Set a user preference for this session.
    
    Each user has their own isolated preferences.
    """
    # Get unique session ID
    session_id = id(ctx.session)
    
    # Initialize session data if needed
    if session_id not in user_preferences:
        user_preferences[session_id] = {}
    
    # Store preference for THIS session only
    user_preferences[session_id][key] = value
    
    return f"Preference '{key}' set to '{value}' for your session"

@mcp.tool()
async def get_preference(
    key: str,
    ctx: Context[ServerSession, None]
) -> str:
    """Get a user preference from this session."""
    session_id = id(ctx.session)
    
    if session_id in user_preferences:
        value = user_preferences[session_id].get(key, "Not set")
        return f"{key}: {value}"
    
    return f"{key}: Not set"

Key points:

  • The Context parameter gives access to session information
  • We use id(ctx.session) as a unique session identifier
  • Each session has isolated data
  • Changes in one session don't affect others

Managing Shared Resources

When multiple clients access shared resources (databases, APIs), you need proper resource pooling. 

import asyncpg
from contextlib import asynccontextmanager

# Create a connection pool (shared across all sessions)
db_pool = None

@asynccontextmanager
async def get_db_connection():
    """Get a database connection from the pool."""
    global db_pool
    
    if db_pool is None:
        # Initialize pool on first use
        db_pool = await asyncpg.create_pool(
            host='localhost',
            database='myapp',
            user='user',
            password='password',
            min_size=5,
            max_size=20
        )
    
    # Get connection from pool
    async with db_pool.acquire() as connection:
        yield connection

@mcp.tool()
async def query_database(sql: str) -> list[dict]:
    """
    Execute a database query.
    
    Uses connection pooling to handle multiple concurrent requests.
    """
    async with get_db_connection() as conn:
        results = await conn.fetch(sql)
        return [dict(row) for row in results]

Connection pooling ensures:

  • Maximum 20 concurrent database connections
  • Minimum 5 connections kept warm for fast response
  • Automatic connection reuse
  • No "too many connections" errors

Rate Limiting Across Sessions

When calling external APIs with rate limits, implement request queuing: 

import asyncio
from collections import deque
from datetime import datetime, timedelta

class RateLimiter:
    """Rate limiter for external API calls."""
    
    def __init__(self, max_requests: int, time_window: int):
        self.max_requests = max_requests
        self.time_window = timedelta(seconds=time_window)
        self.requests = deque()
        self.lock = asyncio.Lock()
    
    async def acquire(self):
        """Wait until we can make a request within rate limit."""
        async with self.lock:
            now = datetime.now()
            
            # Remove old requests outside time window
            while self.requests and self.requests[0] < now - self.time_window:
                self.requests.popleft()
            
            # Wait if we've hit the limit
            if len(self.requests) >= self.max_requests:
                sleep_time = (self.requests[0] + self.time_window - now).total_seconds()
                if sleep_time > 0:
                    await asyncio.sleep(sleep_time)
            
            # Record this request
            self.requests.append(datetime.now())

# Create rate limiter: 10 requests per minute
api_limiter = RateLimiter(max_requests=10, time_window=60)

@mcp.tool()
async def call_external_api(endpoint: str) -> dict:
    """
    Call external API with rate limiting.
    
    Automatically queues requests to respect API limits.
    """
    # Wait for rate limit clearance
    await api_limiter.acquire()
    
    # Make the API call
    async with httpx.AsyncClient() as client:
        response = await client.get(f"https://api.example.com/{endpoint}")
        return response.json()

This rate limiter:

  • Works across ALL client sessions
  • Prevents API quota violations
  • Queues requests fairly
  • Is thread-safe with async locks

File Locking for Concurrent Access

When multiple clients modify shared files, use file locking: 

import asyncio
from pathlib import Path

# File locks dictionary
file_locks: Dict[str, asyncio.Lock] = {}

async def get_file_lock(filepath: str) -> asyncio.Lock:
    """Get or create a lock for a specific file."""
    if filepath not in file_locks:
        file_locks[filepath] = asyncio.Lock()
    return file_locks[filepath]

@mcp.tool()
async def append_to_log(
    filepath: str,
    message: str
) -> dict:
    """
    Safely append to a log file with locking.
    
    Prevents corruption from concurrent writes.
    """
    lock = await get_file_lock(filepath)
    
    async with lock:
        # Only one client can write at a time
        path = Path(filepath)
        async with aiofiles.open(path, 'a') as f:
            await f.write(f"{message}\n")
        
        return {"status": "success", "file": filepath}

This ensures that only one client can write to a file at a time, preventing corruption.

Complete Multi-Client Example

Let's put it all together in a realistic multi-user task management system: 

from mcp.server.fastmcp import FastMCP, Context
from mcp.server.session import ServerSession
from pydantic import BaseModel
from typing import Dict, List
import asyncio

mcp = FastMCP("Task Manager")

# Session-specific task lists
user_tasks: Dict[str, List[dict]] = {}
task_locks: Dict[str, asyncio.Lock] = {}

class Task(BaseModel):
    """Task structure."""
    title: str
    description: str
    priority: int = 1
    completed: bool = False

@mcp.tool()
async def add_task(
    task: Task,
    ctx: Context[ServerSession, None]
) -> dict:
    """
    Add a task to YOUR task list.
    
    Each user has their own isolated task list.
    """
    session_id = str(id(ctx.session))
    
    # Create lock and task list if needed
    if session_id not in task_locks:
        task_locks[session_id] = asyncio.Lock()
        user_tasks[session_id] = []
    
    # Lock prevents race conditions
    async with task_locks[session_id]:
        task_dict = task.dict()
        task_dict['id'] = len(user_tasks[session_id]) + 1
        user_tasks[session_id].append(task_dict)
    
    await ctx.info(f"Task added: {task.title}")
    
    return {
        "status": "created",
        "task_id": task_dict['id'],
        "total_tasks": len(user_tasks[session_id])
    }

@mcp.tool()
async def list_tasks(
    ctx: Context[ServerSession, None]
) -> List[dict]:
    """
    List all YOUR tasks.
    
    You only see tasks from your own session.
    """
    session_id = str(id(ctx.session))
    
    if session_id not in user_tasks:
        return []
    
    async with task_locks[session_id]:
        return user_tasks[session_id].copy()

@mcp.tool()
async def complete_task(
    task_id: int,
    ctx: Context[ServerSession, None]
) -> dict:
    """Mark a task as completed."""
    session_id = str(id(ctx.session))
    
    if session_id not in user_tasks:
        return {"status": "error", "message": "No tasks found"}
    
    async with task_locks[session_id]:
        for task in user_tasks[session_id]:
            if task['id'] == task_id:
                task['completed'] = True
                
                await ctx.report_progress(
                    progress=1.0,
                    total=1.0,
                    message=f"Completed: {task['title']}"
                )
                
                return {"status": "success", "task": task}
    
    return {"status": "error", "message": "Task not found"}

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

This complete example demonstrates:

  • Session isolation (each user sees only their tasks)
  • Concurrent access with locks
  • Progress reporting to clients
  • Logging for debugging
  • Clean error handling

✅ Multi-Client Best Practices

1. Always use session IDs to isolate user data
2. Use connection pools for shared resources (databases, APIs)
3. Implement rate limiting for external API calls
4. Use locks for file operations and critical sections
5. Test with concurrent clients to find race conditions

Testing Multiple Clients

Here's how to test your server with multiple concurrent clients: 

import asyncio
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
from pathlib import Path

async def test_client(client_id: int):
    """Simulate a single client."""
    server_path = Path("task_server.py")
    server_params = StdioServerParameters(
        command="python",
        args=[str(server_path)]
    )
    
    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()
            
            # Add a task
            result = await session.call_tool(
                "add_task",
                arguments={
                    "task": {
                        "title": f"Task from client {client_id}",
                        "description": "Test task",
                        "priority": client_id
                    }
                }
            )
            print(f"Client {client_id}: {result}")
            
            # List tasks
            tasks = await session.call_tool("list_tasks", arguments={})
            print(f"Client {client_id} sees {len(tasks)} tasks")

async def main():
    """Run 5 concurrent clients."""
    tasks = [test_client(i) for i in range(5)]
    await asyncio.gather(*tasks)

if __name__ == "__main__":
    asyncio.run(main())

Run this to verify that each client has isolated data and no race conditions occur.

🎯 Putting It All Together: Complete Real-World Example

Let's build a complete MCP server that demonstrates all concepts: 

from mcp.server.fastmcp import FastMCP, Context
from mcp.server.session import ServerSession
from pydantic import BaseModel, Field, field_validator, EmailStr
from typing import Dict, List, Optional
import asyncio
import asyncpg
from datetime import datetime

# Initialize server
mcp = FastMCP("Enterprise CRM System")

# Shared resources
db_pool: Optional[asyncpg.Pool] = None
session_data: Dict[str, dict] = {}
locks: Dict[str, asyncio.Lock] = {}

# Models
class Customer(BaseModel):
    """Customer record with validation."""
    email: EmailStr = Field(description="Customer email address")
    name: str = Field(min_length=2, max_length=100)
    company: Optional[str] = None
    phone: Optional[str] = Field(None, pattern=r'^\+?[\d\s-()]+$')
    
    @field_validator('name')
    @classmethod
    def validate_name(cls, v):
        if v.lower() in ['test', 'admin']:
            raise ValueError("Invalid name")
        return v

class SearchResult(BaseModel):
    """Structured search results."""
    query: str
    total: int
    customers: List[dict]
    page: int
    per_page: int

# Database initialization
async def get_db():
    """Get database connection from pool."""
    global db_pool
    if db_pool is None:
        db_pool = await asyncpg.create_pool(
            host='localhost',
            database='crm',
            user='user',
            password='password',
            min_size=5,
            max_size=20
        )
    return db_pool

# Tools
@mcp.tool()
async def create_customer(
    customer: Customer,
    ctx: Context[ServerSession, None]
) -> dict:
    """
    Create a new customer with validated data.
    
    Email must be unique. All fields are validated.
    """
    session_id = str(id(ctx.session))
    
    await ctx.info(f"Creating customer: {customer.email}")
    
    pool = await get_db()
    async with pool.acquire() as conn:
        try:
            result = await conn.fetchrow(
                """
                INSERT INTO customers (email, name, company, phone, created_at)
                VALUES ($1, $2, $3, $4, $5)
                RETURNING id
                """,
                customer.email,
                customer.name,
                customer.company,
                customer.phone,
                datetime.now()
            )
            
            customer_id = result['id']
            
            await ctx.report_progress(
                progress=1.0,
                total=1.0,
                message=f"Customer created: {customer.name}"
            )
            
            return {
                "status": "success",
                "customer_id": customer_id,
                "email": customer.email
            }
            
        except asyncpg.UniqueViolationError:
            return {
                "status": "error",
                "message": "Email already exists"
            }

@mcp.tool()
async def search_customers(
    query: str,
    page: int = 1,
    per_page: int = 10,
    ctx: Context[ServerSession, None]
) -> SearchResult:
    """
    Search customers by name, email, or company.
    
    Returns structured, paginated results.
    """
    await ctx.debug(f"Searching: {query}, page {page}")
    
    pool = await get_db()
    async with pool.acquire() as conn:
        # Get total count
        total = await conn.fetchval(
            """
            SELECT COUNT(*) FROM customers
            WHERE name ILIKE $1 OR email ILIKE $1 OR company ILIKE $1
            """,
            f"%{query}%"
        )
        
        # Get paginated results
        offset = (page - 1) * per_page
        rows = await conn.fetch(
            """
            SELECT id, email, name, company, phone, created_at
            FROM customers
            WHERE name ILIKE $1 OR email ILIKE $1 OR company ILIKE $1
            ORDER BY created_at DESC
            LIMIT $2 OFFSET $3
            """,
            f"%{query}%",
            per_page,
            offset
        )
        
        customers = [dict(row) for row in rows]
        
        return SearchResult(
            query=query,
            total=total,
            customers=customers,
            page=page,
            per_page=per_page
        )

@mcp.tool()
async def get_session_stats(
    ctx: Context[ServerSession, None]
) -> dict:
    """
    Get statistics for YOUR session.
    
    Shows how many operations you've performed.
    """
    session_id = str(id(ctx.session))
    
    if session_id not in session_data:
        session_data[session_id] = {
            "operations": 0,
            "started_at": datetime.now().isoformat()
        }
    
    session_data[session_id]["operations"] += 1
    
    return {
        "session_id": session_id[:8],  # Short ID for display
        "operations": session_data[session_id]["operations"],
        "started_at": session_data[session_id]["started_at"]
    }

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

This complete example shows:

  • ✅ Python functions exposed as MCP tools
  • ✅ Comprehensive schema validation with Pydantic
  • ✅ Structured results with proper types
  • ✅ Multiple client handling with sessions
  • ✅ Database connection pooling
  • ✅ Progress reporting and logging
  • ✅ Error handling

🚀 Deployment Considerations

When deploying MCP servers to production, consider:

1. Transport Choice

  • stdio — Best for local tools, simple deployment
  • HTTP + SSE — Best for remote servers, high concurrency
  • HTTP (Streamable) — Modern standard, scales well

2. Monitoring

import logging

logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)

@mcp.tool()
async def monitored_operation(
    data: str,
    ctx: Context[ServerSession, None]
) -> dict:
    """Operation with comprehensive logging."""
    session_id = str(id(ctx.session))
    
    await ctx.info(f"[{session_id[:8]}] Starting operation")
    
    try:
        result = perform_operation(data)
        await ctx.info(f"[{session_id[:8]}] Success")
        return result
    except Exception as e:
        await ctx.error(f"[{session_id[:8]}] Error: {str(e)}")
        raise

3. Security

❌ Security Checklist

1. Validate ALL inputs — Never trust client data
2. Sanitize file paths — Prevent path traversal
3. Use connection pools — Prevent resource exhaustion
4. Implement rate limiting — Prevent abuse
5. Log security events — Track suspicious activity
6. Use HTTPS for remote servers — Encrypt traffic

📚 Summary: From Zero to Hero

Let's recap what you've learned:

Part 1: Exposing Python Functions

  • Use @mcp.tool() decorator to expose functions
  • Add type hints for automatic schema generation
  • Write clear docstrings for AI understanding
  • Use Pydantic models for complex parameters

Part 2: Schema Validation

  • FastMCP validates based on type hints automatically
  • Use Pydantic for advanced validation rules
  • Custom validators prevent security issues
  • Fail fast with clear error messages

Part 3: Structured Results

  • Return Pydantic models for automatic structuring
  • Always add return type annotations
  • MCP provides both structured and text content
  • Use CallToolResult for maximum control

Part 4: Multiple Clients

  • Use Context to access session information
  • Isolate data per session with unique IDs
  • Implement connection pooling for shared resources
  • Use locks for concurrent access to files/data
  • Add rate limiting for external APIs
Labels:

Comments

Post a Comment