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

> Learn to create MCP servers for custom APIs by wrapping endpoints with FastMCP, validating with Pydantic, and exposing via STDIO, HTTP, or SSE for secure LLM agent invocation.

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

---

**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`](https://github.com/ComposioHQ/awesome-codex-skills/blob/main/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`](https://github.com/ComposioHQ/awesome-codex-skills/blob/main/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`](https://github.com/ComposioHQ/awesome-codex-skills/blob/main/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`](https://github.com/ComposioHQ/awesome-codex-skills/blob/main/mcp-builder/reference/python_mcp_server.md):

```python
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:

```python
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:

```python
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:

```python
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:

```python
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`](https://github.com/ComposioHQ/awesome-codex-skills/blob/main/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`](https://github.com/ComposioHQ/awesome-codex-skills/blob/main/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`](https://github.com/ComposioHQ/awesome-codex-skills/blob/main/python_mcp_server.md) guide covers Pydantic validation and FastMCP patterns, while [`node_mcp_server.md`](https://github.com/ComposioHQ/awesome-codex-skills/blob/main/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`](https://github.com/ComposioHQ/awesome-codex-skills/blob/main/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.