1# Python MCP Server Implementation Guide
2 
3## Overview
4 
5This document provides Python-specific best practices and examples for implementing MCP servers using the MCP Python SDK. It covers server setup, tool registration patterns, input validation with Pydantic, error handling, and complete working examples.
6 
7---
8 
9## Quick Reference
10 
11### Key Imports
12```python
13from mcp.server.fastmcp import FastMCP
14from pydantic import BaseModel, Field, field_validator, ConfigDict
15from typing import Optional, List, Dict, Any
16from enum import Enum
17import httpx
18```
19 
20### Server Initialization
21```python
22mcp = FastMCP("service_mcp")
23```
24 
25### Tool Registration Pattern
26```python
27@mcp.tool(name="tool_name", annotations={...})
28async def tool_function(params: InputModel) -> str:
29    # Implementation
30    pass
31```
32 
33---
34 
35## MCP Python SDK and FastMCP
36 
37The official MCP Python SDK provides FastMCP, a high-level framework for building MCP servers. It provides:
38- Automatic description and inputSchema generation from function signatures and docstrings
39- Pydantic model integration for input validation
40- Decorator-based tool registration with `@mcp.tool`
41 
42**For complete SDK documentation, use WebFetch to load:**
43`https://raw.githubusercontent.com/modelcontextprotocol/python-sdk/main/README.md`
44 
45## Server Naming Convention
46 
47Python MCP servers must follow this naming pattern:
48- **Format**: `{service}_mcp` (lowercase with underscores)
49- **Examples**: `github_mcp`, `jira_mcp`, `stripe_mcp`
50 
51The name should be:
52- General (not tied to specific features)
53- Descriptive of the service/API being integrated
54- Easy to infer from the task description
55- Without version numbers or dates
56 
57## Tool Implementation
58 
59### Tool Naming
60 
61Use snake_case for tool names (e.g., "search_users", "create_project", "get_channel_info") with clear, action-oriented names.
62 
63**Avoid Naming Conflicts**: Include the service context to prevent overlaps:
64- Use "slack_send_message" instead of just "send_message"
65- Use "github_create_issue" instead of just "create_issue"
66- Use "asana_list_tasks" instead of just "list_tasks"
67 
68### Tool Structure with FastMCP
69 
70Tools are defined using the `@mcp.tool` decorator with Pydantic models for input validation:
71 
72```python
73from pydantic import BaseModel, Field, ConfigDict
74from mcp.server.fastmcp import FastMCP
75 
76# Initialize the MCP server
77mcp = FastMCP("example_mcp")
78 
79# Define Pydantic model for input validation
80class ServiceToolInput(BaseModel):
81    '''Input model for service tool operation.'''
82    model_config = ConfigDict(
83        str_strip_whitespace=True,  # Auto-strip whitespace from strings
84        validate_assignment=True,    # Validate on assignment
85        extra='forbid'              # Forbid extra fields
86    )
87 
88    param1: str = Field(..., description="First parameter description (e.g., 'user123', 'project-abc')", min_length=1, max_length=100)
89    param2: Optional[int] = Field(default=None, description="Optional integer parameter with constraints", ge=0, le=1000)
90    tags: Optional[List[str]] = Field(default_factory=list, description="List of tags to apply", max_items=10)
91 
92@mcp.tool(
93    name="service_tool_name",
94    annotations={
95        "title": "Human-Readable Tool Title",
96        "readOnlyHint": True,     # Tool does not modify environment
97        "destructiveHint": False,  # Tool does not perform destructive operations
98        "idempotentHint": True,    # Repeated calls have no additional effect
99        "openWorldHint": False     # Tool does not interact with external entities
100    }
101)
102async def service_tool_name(params: ServiceToolInput) -> str:
103    '''Tool description automatically becomes the 'description' field.
104 
105    This tool performs a specific operation on the service. It validates all inputs
106    using the ServiceToolInput Pydantic model before processing.
107 
108    Args:
109        params (ServiceToolInput): Validated input parameters containing:
110            - param1 (str): First parameter description
111            - param2 (Optional[int]): Optional parameter with default
112            - tags (Optional[List[str]]): List of tags
113 
114    Returns:
115        str: JSON-formatted response containing operation results
116    '''
117    # Implementation here
118    pass
119```
120 
121## Pydantic v2 Key Features
122 
123- Use `model_config` instead of nested `Config` class
124- Use `field_validator` instead of deprecated `validator`
125- Use `model_dump()` instead of deprecated `dict()`
126- Validators require `@classmethod` decorator
127- Type hints are required for validator methods
128 
129```python
130from pydantic import BaseModel, Field, field_validator, ConfigDict
131 
132class CreateUserInput(BaseModel):
133    model_config = ConfigDict(
134        str_strip_whitespace=True,
135        validate_assignment=True
136    )
137 
138    name: str = Field(..., description="User's full name", min_length=1, max_length=100)
139    email: str = Field(..., description="User's email address", pattern=r'^[\w\.-]+@[\w\.-]+\.\w+$')
140    age: int = Field(..., description="User's age", ge=0, le=150)
141 
142    @field_validator('email')
143    @classmethod
144    def validate_email(cls, v: str) -> str:
145        if not v.strip():
146            raise ValueError("Email cannot be empty")
147        return v.lower()
148```
149 
150## Response Format Options
151 
152Support multiple output formats for flexibility:
153 
154```python
155from enum import Enum
156 
157class ResponseFormat(str, Enum):
158    '''Output format for tool responses.'''
159    MARKDOWN = "markdown"
160    JSON = "json"
161 
162class UserSearchInput(BaseModel):
163    query: str = Field(..., description="Search query")
164    response_format: ResponseFormat = Field(
165        default=ResponseFormat.MARKDOWN,
166        description="Output format: 'markdown' for human-readable or 'json' for machine-readable"
167    )
168```
169 
170**Markdown format**:
171- Use headers, lists, and formatting for clarity
172- Convert timestamps to human-readable format (e.g., "2024-01-15 10:30:00 UTC" instead of epoch)
173- Show display names with IDs in parentheses (e.g., "@john.doe (U123456)")
174- Omit verbose metadata (e.g., show only one profile image URL, not all sizes)
175- Group related information logically
176 
177**JSON format**:
178- Return complete, structured data suitable for programmatic processing
179- Include all available fields and metadata
180- Use consistent field names and types
181 
182## Pagination Implementation
183 
184For tools that list resources:
185 
186```python
187class ListInput(BaseModel):
188    limit: Optional[int] = Field(default=20, description="Maximum results to return", ge=1, le=100)
189    offset: Optional[int] = Field(default=0, description="Number of results to skip for pagination", ge=0)
190 
191async def list_items(params: ListInput) -> str:
192    # Make API request with pagination
193    data = await api_request(limit=params.limit, offset=params.offset)
194 
195    # Return pagination info
196    response = {
197        "total": data["total"],
198        "count": len(data["items"]),
199        "offset": params.offset,
200        "items": data["items"],
201        "has_more": data["total"] > params.offset + len(data["items"]),
202        "next_offset": params.offset + len(data["items"]) if data["total"] > params.offset + len(data["items"]) else None
203    }
204    return json.dumps(response, indent=2)
205```
206 
207## Error Handling
208 
209Provide clear, actionable error messages:
210 
211```python
212def _handle_api_error(e: Exception) -> str:
213    '''Consistent error formatting across all tools.'''
214    if isinstance(e, httpx.HTTPStatusError):
215        if e.response.status_code == 404:
216            return "Error: Resource not found. Please check the ID is correct."
217        elif e.response.status_code == 403:
218            return "Error: Permission denied. You don't have access to this resource."
219        elif e.response.status_code == 429:
220            return "Error: Rate limit exceeded. Please wait before making more requests."
221        return f"Error: API request failed with status {e.response.status_code}"
222    elif isinstance(e, httpx.TimeoutException):
223        return "Error: Request timed out. Please try again."
224    return f"Error: Unexpected error occurred: {type(e).__name__}"
225```
226 
227## Shared Utilities
228 
229Extract common functionality into reusable functions:
230 
231```python
232# Shared API request function
233async def _make_api_request(endpoint: str, method: str = "GET", **kwargs) -> dict:
234    '''Reusable function for all API calls.'''
235    async with httpx.AsyncClient() as client:
236        response = await client.request(
237            method,
238            f"{API_BASE_URL}/{endpoint}",
239            timeout=30.0,
240            **kwargs
241        )
242        response.raise_for_status()
243        return response.json()
244```
245 
246## Async/Await Best Practices
247 
248Always use async/await for network requests and I/O operations:
249 
250```python
251# Good: Async network request
252async def fetch_data(resource_id: str) -> dict:
253    async with httpx.AsyncClient() as client:
254        response = await client.get(f"{API_URL}/resource/{resource_id}")
255        response.raise_for_status()
256        return response.json()
257 
258# Bad: Synchronous request
259def fetch_data(resource_id: str) -> dict:
260    response = requests.get(f"{API_URL}/resource/{resource_id}")  # Blocks
261    return response.json()
262```
263 
264## Type Hints
265 
266Use type hints throughout:
267 
268```python
269from typing import Optional, List, Dict, Any
270 
271async def get_user(user_id: str) -> Dict[str, Any]:
272    data = await fetch_user(user_id)
273    return {"id": data["id"], "name": data["name"]}
274```
275 
276## Tool Docstrings
277 
278Every tool must have comprehensive docstrings with explicit type information:
279 
280```python
281async def search_users(params: UserSearchInput) -> str:
282    '''
283    Search for users in the Example system by name, email, or team.
284 
285    This tool searches across all user profiles in the Example platform,
286    supporting partial matches and various search filters. It does NOT
287    create or modify users, only searches existing ones.
288 
289    Args:
290        params (UserSearchInput): Validated input parameters containing:
291            - query (str): Search string to match against names/emails (e.g., "john", "@example.com", "team:marketing")
292            - limit (Optional[int]): Maximum results to return, between 1-100 (default: 20)
293            - offset (Optional[int]): Number of results to skip for pagination (default: 0)
294 
295    Returns:
296        str: JSON-formatted string containing search results with the following schema:
297 
298        Success response:
299        {
300            "total": int,           # Total number of matches found
301            "count": int,           # Number of results in this response
302            "offset": int,          # Current pagination offset
303            "users": [
304                {
305                    "id": str,      # User ID (e.g., "U123456789")
306                    "name": str,    # Full name (e.g., "John Doe")
307                    "email": str,   # Email address (e.g., "[email protected]")
308                    "team": str     # Team name (e.g., "Marketing") - optional
309                }
310            ]
311        }
312 
313        Error response:
314        "Error: <error message>" or "No users found matching '<query>'"
315 
316    Examples:
317        - Use when: "Find all marketing team members" -> params with query="team:marketing"
318        - Use when: "Search for John's account" -> params with query="john"
319        - Don't use when: You need to create a user (use example_create_user instead)
320        - Don't use when: You have a user ID and need full details (use example_get_user instead)
321 
322    Error Handling:
323        - Input validation errors are handled by Pydantic model
324        - Returns "Error: Rate limit exceeded" if too many requests (429 status)
325        - Returns "Error: Invalid API authentication" if API key is invalid (401 status)
326        - Returns formatted list of results or "No users found matching 'query'"
327    '''
328```
329 
330## Complete Example
331 
332See below for a complete Python MCP server example:
333 
334```python
335#!/usr/bin/env python3
336'''
337MCP Server for Example Service.
338 
339This server provides tools to interact with Example API, including user search,
340project management, and data export capabilities.
341'''
342 
343from typing import Optional, List, Dict, Any
344from enum import Enum
345import httpx
346from pydantic import BaseModel, Field, field_validator, ConfigDict
347from mcp.server.fastmcp import FastMCP
348 
349# Initialize the MCP server
350mcp = FastMCP("example_mcp")
351 
352# Constants
353API_BASE_URL = "https://api.example.com/v1"
354 
355# Enums
356class ResponseFormat(str, Enum):
357    '''Output format for tool responses.'''
358    MARKDOWN = "markdown"
359    JSON = "json"
360 
361# Pydantic Models for Input Validation
362class UserSearchInput(BaseModel):
363    '''Input model for user search operations.'''
364    model_config = ConfigDict(
365        str_strip_whitespace=True,
366        validate_assignment=True
367    )
368 
369    query: str = Field(..., description="Search string to match against names/emails", min_length=2, max_length=200)
370    limit: Optional[int] = Field(default=20, description="Maximum results to return", ge=1, le=100)
371    offset: Optional[int] = Field(default=0, description="Number of results to skip for pagination", ge=0)
372    response_format: ResponseFormat = Field(default=ResponseFormat.MARKDOWN, description="Output format")
373 
374    @field_validator('query')
375    @classmethod
376    def validate_query(cls, v: str) -> str:
377        if not v.strip():
378            raise ValueError("Query cannot be empty or whitespace only")
379        return v.strip()
380 
381# Shared utility functions
382async def _make_api_request(endpoint: str, method: str = "GET", **kwargs) -> dict:
383    '''Reusable function for all API calls.'''
384    async with httpx.AsyncClient() as client:
385        response = await client.request(
386            method,
387            f"{API_BASE_URL}/{endpoint}",
388            timeout=30.0,
389            **kwargs
390        )
391        response.raise_for_status()
392        return response.json()
393 
394def _handle_api_error(e: Exception) -> str:
395    '''Consistent error formatting across all tools.'''
396    if isinstance(e, httpx.HTTPStatusError):
397        if e.response.status_code == 404:
398            return "Error: Resource not found. Please check the ID is correct."
399        elif e.response.status_code == 403:
400            return "Error: Permission denied. You don't have access to this resource."
401        elif e.response.status_code == 429:
402            return "Error: Rate limit exceeded. Please wait before making more requests."
403        return f"Error: API request failed with status {e.response.status_code}"
404    elif isinstance(e, httpx.TimeoutException):
405        return "Error: Request timed out. Please try again."
406    return f"Error: Unexpected error occurred: {type(e).__name__}"
407 
408# Tool definitions
409@mcp.tool(
410    name="example_search_users",
411    annotations={
412        "title": "Search Example Users",
413        "readOnlyHint": True,
414        "destructiveHint": False,
415        "idempotentHint": True,
416        "openWorldHint": True
417    }
418)
419async def example_search_users(params: UserSearchInput) -> str:
420    '''Search for users in the Example system by name, email, or team.
421 
422    [Full docstring as shown above]
423    '''
424    try:
425        # Make API request using validated parameters
426        data = await _make_api_request(
427            "users/search",
428            params={
429                "q": params.query,
430                "limit": params.limit,
431                "offset": params.offset
432            }
433        )
434 
435        users = data.get("users", [])
436        total = data.get("total", 0)
437 
438        if not users:
439            return f"No users found matching '{params.query}'"
440 
441        # Format response based on requested format
442        if params.response_format == ResponseFormat.MARKDOWN:
443            lines = [f"# User Search Results: '{params.query}'", ""]
444            lines.append(f"Found {total} users (showing {len(users)})")
445            lines.append("")
446 
447            for user in users:
448                lines.append(f"## {user['name']} ({user['id']})")
449                lines.append(f"- **Email**: {user['email']}")
450                if user.get('team'):
451                    lines.append(f"- **Team**: {user['team']}")
452                lines.append("")
453 
454            return "\n".join(lines)
455 
456        else:
457            # Machine-readable JSON format
458            import json
459            response = {
460                "total": total,
461                "count": len(users),
462                "offset": params.offset,
463                "users": users
464            }
465            return json.dumps(response, indent=2)
466 
467    except Exception as e:
468        return _handle_api_error(e)
469 
470if __name__ == "__main__":
471    mcp.run()
472```
473 
474---
475 
476## Advanced FastMCP Features
477 
478### Context Parameter Injection
479 
480FastMCP can automatically inject a `Context` parameter into tools for advanced capabilities like logging, progress reporting, resource reading, and user interaction:
481 
482```python
483from mcp.server.fastmcp import FastMCP, Context
484 
485mcp = FastMCP("example_mcp")
486 
487@mcp.tool()
488async def advanced_search(query: str, ctx: Context) -> str:
489    '''Advanced tool with context access for logging and progress.'''
490 
491    # Report progress for long operations
492    await ctx.report_progress(0.25, "Starting search...")
493 
494    # Log information for debugging
495    await ctx.log_info("Processing query", {"query": query, "timestamp": datetime.now()})
496 
497    # Perform search
498    results = await search_api(query)
499    await ctx.report_progress(0.75, "Formatting results...")
500 
501    # Access server configuration
502    server_name = ctx.fastmcp.name
503 
504    return format_results(results)
505 
506@mcp.tool()
507async def interactive_tool(resource_id: str, ctx: Context) -> str:
508    '''Tool that can request additional input from users.'''
509 
510    # Request sensitive information when needed
511    api_key = await ctx.elicit(
512        prompt="Please provide your API key:",
513        input_type="password"
514    )
515 
516    # Use the provided key
517    return await api_call(resource_id, api_key)
518```
519 
520**Context capabilities:**
521- `ctx.report_progress(progress, message)` - Report progress for long operations
522- `ctx.log_info(message, data)` / `ctx.log_error()` / `ctx.log_debug()` - Logging
523- `ctx.elicit(prompt, input_type)` - Request input from users
524- `ctx.fastmcp.name` - Access server configuration
525- `ctx.read_resource(uri)` - Read MCP resources
526 
527### Resource Registration
528 
529Expose data as resources for efficient, template-based access:
530 
531```python
532@mcp.resource("file://documents/{name}")
533async def get_document(name: str) -> str:
534    '''Expose documents as MCP resources.
535 
536    Resources are useful for static or semi-static data that doesn't
537    require complex parameters. They use URI templates for flexible access.
538    '''
539    document_path = f"./docs/{name}"
540    with open(document_path, "r") as f:
541        return f.read()
542 
543@mcp.resource("config://settings/{key}")
544async def get_setting(key: str, ctx: Context) -> str:
545    '''Expose configuration as resources with context.'''
546    settings = await load_settings()
547    return json.dumps(settings.get(key, {}))
548```
549 
550**When to use Resources vs Tools:**
551- **Resources**: For data access with simple parameters (URI templates)
552- **Tools**: For complex operations with validation and business logic
553 
554### Structured Output Types
555 
556FastMCP supports multiple return types beyond strings:
557 
558```python
559from typing import TypedDict
560from dataclasses import dataclass
561from pydantic import BaseModel
562 
563# TypedDict for structured returns
564class UserData(TypedDict):
565    id: str
566    name: str
567    email: str
568 
569@mcp.tool()
570async def get_user_typed(user_id: str) -> UserData:
571    '''Returns structured data - FastMCP handles serialization.'''
572    return {"id": user_id, "name": "John Doe", "email": "[email protected]"}
573 
574# Pydantic models for complex validation
575class DetailedUser(BaseModel):
576    id: str
577    name: str
578    email: str
579    created_at: datetime
580    metadata: Dict[str, Any]
581 
582@mcp.tool()
583async def get_user_detailed(user_id: str) -> DetailedUser:
584    '''Returns Pydantic model - automatically generates schema.'''
585    user = await fetch_user(user_id)
586    return DetailedUser(**user)
587```
588 
589### Lifespan Management
590 
591Initialize resources that persist across requests:
592 
593```python
594from contextlib import asynccontextmanager
595 
596@asynccontextmanager
597async def app_lifespan():
598    '''Manage resources that live for the server's lifetime.'''
599    # Initialize connections, load config, etc.
600    db = await connect_to_database()
601    config = load_configuration()
602 
603    # Make available to all tools
604    yield {"db": db, "config": config}
605 
606    # Cleanup on shutdown
607    await db.close()
608 
609mcp = FastMCP("example_mcp", lifespan=app_lifespan)
610 
611@mcp.tool()
612async def query_data(query: str, ctx: Context) -> str:
613    '''Access lifespan resources through context.'''
614    db = ctx.request_context.lifespan_state["db"]
615    results = await db.query(query)
616    return format_results(results)
617```
618 
619### Transport Options
620 
621FastMCP supports two main transport mechanisms:
622 
623```python
624# stdio transport (for local tools) - default
625if __name__ == "__main__":
626    mcp.run()
627 
628# Streamable HTTP transport (for remote servers)
629if __name__ == "__main__":
630    mcp.run(transport="streamable_http", port=8000)
631```
632 
633**Transport selection:**
634- **stdio**: Command-line tools, local integrations, subprocess execution
635- **Streamable HTTP**: Web services, remote access, multiple clients
636 
637---
638 
639## Code Best Practices
640 
641### Code Composability and Reusability
642 
643Your implementation MUST prioritize composability and code reuse:
644 
6451. **Extract Common Functionality**:
646   - Create reusable helper functions for operations used across multiple tools
647   - Build shared API clients for HTTP requests instead of duplicating code
648   - Centralize error handling logic in utility functions
649   - Extract business logic into dedicated functions that can be composed
650   - Extract shared markdown or JSON field selection & formatting functionality
651 
6522. **Avoid Duplication**:
653   - NEVER copy-paste similar code between tools
654   - If you find yourself writing similar logic twice, extract it into a function
655   - Common operations like pagination, filtering, field selection, and formatting should be shared
656   - Authentication/authorization logic should be centralized
657 
658### Python-Specific Best Practices
659 
6601. **Use Type Hints**: Always include type annotations for function parameters and return values
6612. **Pydantic Models**: Define clear Pydantic models for all input validation
6623. **Avoid Manual Validation**: Let Pydantic handle input validation with constraints
6634. **Proper Imports**: Group imports (standard library, third-party, local)
6645. **Error Handling**: Use specific exception types (httpx.HTTPStatusError, not generic Exception)
6656. **Async Context Managers**: Use `async with` for resources that need cleanup
6667. **Constants**: Define module-level constants in UPPER_CASE
667 
668## Quality Checklist
669 
670Before finalizing your Python MCP server implementation, ensure:
671 
672### Strategic Design
673- [ ] Tools enable complete workflows, not just API endpoint wrappers
674- [ ] Tool names reflect natural task subdivisions
675- [ ] Response formats optimize for agent context efficiency
676- [ ] Human-readable identifiers used where appropriate
677- [ ] Error messages guide agents toward correct usage
678 
679### Implementation Quality
680- [ ] FOCUSED IMPLEMENTATION: Most important and valuable tools implemented
681- [ ] All tools have descriptive names and documentation
682- [ ] Return types are consistent across similar operations
683- [ ] Error handling is implemented for all external calls
684- [ ] Server name follows format: `{service}_mcp`
685- [ ] All network operations use async/await
686- [ ] Common functionality is extracted into reusable functions
687- [ ] Error messages are clear, actionable, and educational
688- [ ] Outputs are properly validated and formatted
689 
690### Tool Configuration
691- [ ] All tools implement 'name' and 'annotations' in the decorator
692- [ ] Annotations correctly set (readOnlyHint, destructiveHint, idempotentHint, openWorldHint)
693- [ ] All tools use Pydantic BaseModel for input validation with Field() definitions
694- [ ] All Pydantic Fields have explicit types and descriptions with constraints
695- [ ] All tools have comprehensive docstrings with explicit input/output types
696- [ ] Docstrings include complete schema structure for dict/JSON returns
697- [ ] Pydantic models handle input validation (no manual validation needed)
698 
699### Advanced Features (where applicable)
700- [ ] Context injection used for logging, progress, or elicitation
701- [ ] Resources registered for appropriate data endpoints
702- [ ] Lifespan management implemented for persistent connections
703- [ ] Structured output types used (TypedDict, Pydantic models)
704- [ ] Appropriate transport configured (stdio or streamable HTTP)
705 
706### Code Quality
707- [ ] File includes proper imports including Pydantic imports
708- [ ] Pagination is properly implemented where applicable
709- [ ] Filtering options are provided for potentially large result sets
710- [ ] All async functions are properly defined with `async def`
711- [ ] HTTP client usage follows async patterns with proper context managers
712- [ ] Type hints are used throughout the code
713- [ ] Constants are defined at module level in UPPER_CASE
714 
715### Testing
716- [ ] Server runs successfully: `python your_server.py --help`
717- [ ] All imports resolve correctly
718- [ ] Sample tool calls work as expected
719- [ ] Error scenarios handled gracefully