How to Create MCP Servers for Custom APIs: A Production-Ready Guide

Creating an MCP server for custom APIs involves wrapping your REST or GraphQL endpoints as tools using the FastMCP framework, validating inputs with Pydantic v2 models, and exposing them via STDIO, HTTP, or SSE transports so LLM agents can invoke them safely.

The ComposioHQ/awesome-codex-skills repository provides a complete framework for building Model Context Protocol (MCP) servers that expose any external API as callable tools for AI agents. By implementing the patterns found in mcp-builder/reference/python_mcp_server.md, you can create MCP servers for custom APIs that handle input validation, pagination, error handling, and multiple transport layers automatically.

Architecture and Core Design Patterns

The FastMCP framework provides a high-level wrapper that automatically generates OpenAI-compatible function specifications from Python signatures and docstrings. According to the source code in mcp-builder/reference/python_mcp_server.md, every production-ready MCP server follows these architectural patterns.

FastMCP Server Instance

Instantiate the server with a name following the {service}_mcp convention. The FastMCP class handles protocol compliance, tool discovery, and transport management automatically.

Tool Registration with @mcp.tool

Each API endpoint becomes a callable tool through the decorator pattern. The decorator requires:

• name – Snake-case identifier prefixed with the service name (e.g., github_create_issue)
• annotations – Metadata flags including readOnlyHint, destructiveHint, idempotentHint, and openWorldHint that guide LLMs on side effects

Input Validation Using Pydantic v2

Define strict input models using BaseModel with ConfigDict to strip whitespace, forbid extra fields, and enable validation on assignment. This ensures type safety before any HTTP request executes.

Response Format Standardization

Tools return either Markdown for human readability or JSON for programmatic consumption. A ResponseFormat enum selects the output style, while a module-level CHARACTER_LIMIT constant caps response size to prevent context window overflow.

Error Handling and Shared Utilities

Centralize HTTP logic in _make_api_request using httpx.AsyncClient with consistent timeouts. Convert exceptions into actionable messages through _handle_api_error, which specifically handles HTTP status codes (404, 403, 429) and timeout scenarios.

Resources vs. Tools

Expose static data (documentation, schemas) via @mcp.resource URIs. Reserve @mcp.tool decorators for state-changing operations and dynamic queries.

Lifespan Management

Use an async context manager (app_lifespan) to initialize long-lived connections such as database pools or authentication token caches. Inject these resources into tool handlers via ctx.request_context.lifespan_state.

Transport Layer Flexibility

The same server supports three transport modes:

• STDIO (default) for CLI integration
• HTTP (transport="streamable_http") for service-to-service communication
• SSE (transport="sse") for real-time streaming

Step-by-Step Implementation Guide

Follow these concrete steps to implement an MCP server for your custom API:

1. Create a new Python module (e.g., myapi_mcp.py)
2. Import FastMCP from mcp.server.fastmcp, plus httpx and Pydantic classes
3. Instantiate the server: mcp = FastMCP("myapi_mcp")
4. Define constants: API_BASE_URL and CHARACTER_LIMIT (typically 25,000)
5. Write Pydantic input models for each tool endpoint, including pagination parameters
6. Implement _make_api_request and _handle_api_error helpers
7. Register tools using @mcp.tool(name="...", annotations={...})
8. Implement tool logic using validated models and the response format enum
9. Optionally add @mcp.resource handlers for static lookups
10. Optionally define an app_lifespan context manager for persistent connections
11. Select transport and run: mcp.run() or mcp.run(transport="streamable_http", port=8000)

Complete Code Examples

Minimal Server Skeleton

Start with this foundation from mcp-builder/reference/python_mcp_server.md:

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

# Server instantiation following {service}_mcp naming convention

mcp = FastMCP("myapi_mcp")

# Configuration constants

API_BASE_URL = "https://api.myservice.com/v1"
CHARACTER_LIMIT = 25000

# Shared HTTP utility

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

# Centralized error handler

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

Input Models with Validation

Define strict schemas using Pydantic v2 features:

from enum import Enum

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

class ListProjectsInput(BaseModel):
    """Parameters for listing projects with pagination."""
    model_config = ConfigDict(
        str_strip_whitespace=True, 
        validate_assignment=True, 
        extra="forbid"
    )
    
    limit: int = Field(default=20, ge=1, le=100, description="Maximum items to return")
    offset: int = Field(default=0, ge=0, description="Items to skip")
    response_format: ResponseFormat = Field(
        default=ResponseFormat.JSON, 
        description="Output type"
    )
    query: str = Field(..., description="Search query string")

Full Tool Implementation

This example demonstrates the complete pattern from the reference implementation:

import json
from mcp.server.fastmcp import Context

@mcp.tool(
    name="myapi_search_users",
    annotations={
        "title": "Search Users",
        "readOnlyHint": True,
        "destructiveHint": False,
        "idempotentHint": True,
        "openWorldHint": True,
    },
)
async def myapi_search_users(params: ListProjectsInput, ctx: Context) -> str:
    """
    Search for users in MyAPI.
    
    Returns either a Markdown summary or JSON payload based on response_format.
    """
    try:
        data = await _make_api_request(
            "users/search",
            params={
                "q": params.query, 
                "limit": params.limit, 
                "offset": params.offset
            },
        )
        
        users = data.get("users", [])
        total = data.get("total", 0)

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

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

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

                lines.append(f"- **Email**: {u['email']}")
                if u.get("team"):
                    lines.append(f"- **Team**: {u['team']}")
                lines.append("")
                
            return "\n".join(lines[:CHARACTER_LIMIT])

        # JSON output path

        return json.dumps({
            "total": total, 
            "count": len(users), 
            "offset": params.offset, 
            "users": users
        }, indent=2)

    except Exception as e:
        return _handle_api_error(e)

Lifespan Management for Database Connections

Implement persistent connections using the optional lifespan context manager:

from contextlib import asynccontextmanager
import asyncpg

@asynccontextmanager
async def app_lifespan():
    """Initialize long-lived resources for the MCP server."""
    db = await asyncpg.create_pool(dsn="postgres://user:pwd@db:5432/mydb")
    yield {"db": db}
    await db.close()

# Re-instantiate with lifespan support

mcp = FastMCP("myapi_mcp", lifespan=app_lifespan)

@mcp.tool()
async def get_project(project_id: str, ctx: Context) -> str:
    """Fetch project from database using lifespan-managed connection."""
    db = ctx.request_context.lifespan_state["db"]
    row = await db.fetchrow("SELECT * FROM projects WHERE id = $1", project_id)
    return json.dumps(dict(row), indent=2) if row else "Project not found"

Running the Server

Choose your transport layer based on deployment requirements:

if __name__ == "__main__":
    # STDIO transport (default) for CLI integration with Claude Desktop

    mcp.run()
    
    # Alternative: HTTP transport for service mesh deployment

    # mcp.run(transport="streamable_http", port=8000)

    
    # Alternative: SSE transport for real-time streaming

    # mcp.run(transport="sse", port=8000)

MCP Server Best Practices

The mcp_best_practices.md file in the repository specifies these production requirements:

• Name tools with snake_case and service prefixes (e.g., stripe_create_customer) to avoid collisions across multiple MCP servers
• Set accurate annotations on every tool so LLMs understand read-only vs. destructive operations
• Implement pagination using limit and offset fields with maximum value constraints
• Enforce character limits to prevent token overflow in LLM context windows
• Centralize utilities to avoid duplicating HTTP logic or error formatting across tools
• Use context injection (ctx: Context) when tools require logging, progress reporting, or user elicitation via ctx.elicit
• Validate inputs strictly with Pydantic before executing any external API calls
• Test exhaustively using the evaluation script patterns found in evaluation.md

Summary

Creating MCP servers for custom APIs requires following the FastMCP architectural patterns established in the ComposioHQ/awesome-codex-skills repository:

• Instantiate a FastMCP server with the {service}_mcp naming convention
• Register tools using @mcp.tool with descriptive names and accurate annotations
• Validate inputs using Pydantic v2 models with strict configuration
• Handle responses consistently using the ResponseFormat enum and character limits
• Centralize logic in _make_api_request and _handle_api_error helpers
• Manage state optionally through app_lifespan context managers
• Select transport based on deployment needs (STDIO, HTTP, or SSE)

Frequently Asked Questions

What is the Model Context Protocol (MCP)?

MCP is an open protocol that standardizes how applications provide context to Large Language Models. When you create MCP servers for custom APIs, you expose your endpoints as tools that any MCP-compatible client (like Claude Desktop) can discover and invoke safely with structured inputs and outputs.

How do I handle authentication for my custom API?

Store authentication credentials in environment variables or secure secret stores, then reference them in your _make_api_request helper. For token-based auth that requires refresh, implement a lifespan manager that maintains an auth token pool in app_lifespan and injects it via ctx.request_context.lifespan_state.

Should I use Python or Node.js for my MCP server?

Both are fully supported. The python_mcp_server.md guide covers Pydantic validation and FastMCP patterns, while node_mcp_server.md provides equivalent functionality using Zod for schema validation and the TypeScript SDK. Choose based on your existing codebase and team expertise.

How do I test my MCP server before deployment?

Use the evaluation patterns from mcp-builder/reference/evaluation.md to create a test suite that sends sample requests to your tools and validates the responses against expected schemas. Test with all three transport modes (STDIO, HTTP, SSE) to ensure consistent behavior across deployment scenarios.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}