MCP API Server: Wrap REST APIs as MCP Tools for AI Agents
TL;DR — MCP API server wraps REST APIs as MCP tools with FastMCP + httpx. WorkOS: "Three files: server (MCP surface), client (REST API wrapper), auth (bearer token verification). FastMCP handles JSON-RPC, schema, transport. streamable-http for remote with OAuth middleware." MCP Template: "Production-ready template: api_client.py wraps REST API, server.py defines tools. Env vars for API keys, httpx timeouts, JSON string returns, Dockerfile." Medium: "Slack, Notion, Jira MCP servers. Tools are like POST endpoints — they do something. Same structure for Salesforce, internal APIs, databases." DEV Community: "Type parameters, write docstrings, add caching, handle errors, keep tools focused. 17,000+ MCP servers but most undocumented." FastMCP: "Standard framework. 1M downloads/day. Powers 70% of MCP servers. Best practices built in." Learn more with MCP guide, build MCP server, MCP tools, and MCP database server.
WorkOS provides the definitive guide: "This is a practical guide to building an MCP server that wraps an existing REST API. By the end, you'll have a working server in Python that exposes a real product surface to an LLM agent through tools, resources, and prompts, with proper error handling, OAuth-based authentication, and a path from local development to a remote production deployment."
Medium describes the pattern: "Tools are like POST endpoints — they do something and may have side effects. The LLM calls them when it needs to act. Slack, Notion, Jira, Salesforce, your internal APIs, your proprietary databases, they all follow the same structure."
MCP API Server Architecture
'Send Slack message to #dev'"] Approval["User Approval
before tool execution"] end subgraph MCPServer["MCP API Server (FastMCP)"] Tools["MCP Tools
@mcp.tool() decorated functions
schema from type hints"] Prompts["MCP Prompts
reusable templates
e.g., sprint_review"] Resources["MCP Resources
static data
e.g., API docs"] end subgraph Client["API Client Layer"] APIClient["API Client Class
httpx.AsyncClient
base_url, auth headers
timeout=10.0"] Methods["Methods
get, post, patch, delete
raise_for_status"] Cache["TTL Cache
reduce API calls
cache frequent requests"] end subgraph Auth["Authentication"] StdioAuth["stdio: env vars
API_KEY from os.environ"] HTTPAuth["HTTP: OAuth middleware
bearer token verification
WWW-Authenticate header"] end subgraph APIs["External REST APIs"] Slack["Slack API
slack-sdk
messages, channels"] Notion["Notion API
notion-client
pages, databases"] Jira["Jira API
REST v3 via httpx
issues, projects"] Weather["Weather API
wttr.in, OpenWeatherMap
current conditions"] Custom["Custom API
any REST endpoint
your internal APIs"] end Query --> Approval Approval --> Tools Tools --> APIClient APIClient --> Methods Methods --> Cache StdioAuth --> APIClient HTTPAuth --> MCPServer Cache --> Slack Cache --> Notion Cache --> Jira Cache --> Weather Cache --> Custom Prompts --> Tools Resources --> Tools
API Integration Patterns
| Service | SDK/Library | Key Tools | Transport | Auth |
|---|---|---|---|---|
| Slack | slack-sdk | get_messages, send_message, list_channels | stdio | env: SLACK_TOKEN |
| Notion | notion-client | search_pages, create_page, get_page | stdio | env: NOTION_TOKEN |
| Jira | httpx (REST v3) | search_issues, create_issue, update_issue | stdio | env: JIRA_TOKEN |
| Weather | httpx (wttr.in) | get_weather, get_forecast | stdio | none (free API) |
| Custom | httpx | any REST endpoint | stdio or HTTP | env or OAuth |
Tool Design Principles
| Principle | What | Why |
|---|---|---|
| One tool = one job | Each tool does one thing | AI can decide when to call it |
| Type hints | str, int, bool, list, dict | Auto-validation and schema |
| Docstrings | Detailed description with args | AI reads to decide tool calls |
| JSON returns | Return JSON strings for structured data | AI can parse and use |
| Error handling | Return error dicts, not stack traces | AI can understand and retry |
| Timeouts | httpx timeout=10-15s | Prevent hanging |
| Caching | TTL cache for frequent calls | Reduce API rate limits |
| Env vars | API keys in os.environ | Never hardcode secrets |
Implementation
from dataclasses import dataclass
from typing import Optional
from enum import Enum
class APIService(Enum):
SLACK = "slack"
NOTION = "notion"
JIRA = "jira"
WEATHER = "weather"
CUSTOM = "custom"
@dataclass
class MCPAPIServerGuide:
"""MCP API server implementation guide."""
def get_install(self) -> str:
"""Installation for MCP API server."""
return (
"# Core MCP SDK\n"
"pip install 'mcp[cli]' httpx pydantic\n"
"\n"
"# For OAuth (remote deployment)\n"
"pip install python-jose\n"
"\n"
"# Service-specific SDKs\n"
"pip install slack-sdk # Slack\n"
"pip install notion-client # Notion\n"
"# Jira uses httpx (REST API)\n"
"\n"
"# Or use uv\n"
"uv add 'mcp[cli]' httpx pydantic"
)
def get_api_client_pattern(self) -> str:
"""API client class pattern."""
return (
"import os\n"
"import httpx\n"
"from typing import Any\n"
"\n"
"class APIClient:\n"
" '''Wrapper for external REST API.'''\n"
"\n"
" def __init__(self):\n"
" self.base_url = os.environ[\n"
" 'API_BASE_URL']\n"
" self.api_key = os.environ.get(\n"
" 'API_KEY')\n"
" self._client = httpx.AsyncClient(\n"
" base_url=self.base_url,\n"
" headers={\n"
" 'Authorization':\n"
" f'Bearer {self.api_key}'\n"
" } if self.api_key else {},\n"
" timeout=10.0,\n"
" )\n"
"\n"
" async def get(\n"
" self, path: str, **params\n"
" ) -> Any:\n"
" r = await self._client.get(\n"
" path, params=params)\n"
" r.raise_for_status()\n"
" return r.json()\n"
"\n"
" async def post(\n"
" self, path: str,\n"
" json: dict\n"
" ) -> Any:\n"
" r = await self._client.post(\n"
" path, json=json)\n"
" r.raise_for_status()\n"
" return r.json()\n"
"\n"
" async def patch(\n"
" self, path: str,\n"
" json: dict\n"
" ) -> Any:\n"
" r = await self._client.patch(\n"
" path, json=json)\n"
" r.raise_for_status()\n"
" return r.json()\n"
"\n"
" async def delete(\n"
" self, path: str\n"
" ) -> Any:\n"
" r = await self._client.delete(path)\n"
" r.raise_for_status()\n"
" return r.json()\n"
"\n"
" async def aclose(self):\n"
" await self._client.aclose()"
)
def get_server_code(self) -> str:
"""MCP server with API tools."""
return (
"import os\n"
"import json\n"
"import httpx\n"
"from mcp.server.fastmcp import FastMCP\n"
"\n"
"mcp = FastMCP(\n"
" 'api-server',\n"
" instructions=(\n"
" 'This server provides API '\n"
" 'integration tools.'\n"
" ),\n"
")\n"
"\n"
"client = httpx.AsyncClient(\n"
" base_url=os.environ.get(\n"
" 'API_BASE_URL',\n"
" 'https://api.example.com'),\n"
" headers={\n"
" 'Authorization':\n"
" f'Bearer {os.environ.get(\n"
" \"API_KEY\", \"\")}'\n"
" },\n"
" timeout=10.0,\n"
")\n"
"\n"
"@mcp.tool()\n"
"async def search_items(\n"
" query: str, limit: int = 10\n"
") -> str:\n"
" '''Search items via API.\n"
"\n"
" Args:\n"
" query: Search query string\n"
" limit: Max results (1-100)\n"
" '''\n"
" try:\n"
" r = await client.get(\n"
" '/search',\n"
" params={'q': query,\n"
" 'limit': limit})\n"
" r.raise_for_status()\n"
" return json.dumps(\n"
" r.json(), indent=2)\n"
" except httpx.HTTPStatusError as e:\n"
" return json.dumps({\n"
" 'error': 'API error',\n"
" 'status': e.response\n"
" .status_code\n"
" })\n"
" except httpx.TimeoutException:\n"
" return json.dumps({\n"
" 'error': 'Timeout'\n"
" })\n"
"\n"
"@mcp.tool()\n"
"async def get_item(\n"
" item_id: str\n"
") -> str:\n"
" '''Get a single item by ID.\n"
"\n"
" Args:\n"
" item_id: The item identifier\n"
" '''\n"
" r = await client.get(\n"
" f'/items/{item_id}')\n"
" r.raise_for_status()\n"
" return json.dumps(\n"
" r.json(), indent=2)\n"
"\n"
"@mcp.tool()\n"
"async def create_item(\n"
" name: str,\n"
" description: str = ''\n"
") -> str:\n"
" '''Create a new item.\n"
"\n"
" Args:\n"
" name: Item name\n"
" description: Item description\n"
" '''\n"
" r = await client.post(\n"
" '/items',\n"
" json={'name': name,\n"
" 'description': description})\n"
" r.raise_for_status()\n"
" return json.dumps(\n"
" r.json(), indent=2)\n"
"\n"
"@mcp.prompt()\n"
"def summarize_items(\n"
" query: str\n"
") -> str:\n"
" '''Summarize search results.'''\n"
" return (\n"
" f'Search for items matching '\n"
" f'\"{query}\" and provide a '\n"
" f'summary of the results.'\n"
" )\n"
"\n"
"if __name__ == '__main__':\n"
" mcp.run(transport='stdio')"
)
def get_weather_example(self) -> str:
"""Weather API MCP server example."""
return (
"import httpx\n"
"from mcp.server.fastmcp import FastMCP\n"
"\n"
"mcp = FastMCP('weather-server')\n"
"\n"
"@mcp.tool()\n"
"async def get_weather(\n"
" city: str\n"
") -> dict:\n"
" '''Get current weather for a city.\n"
"\n"
" Args:\n"
" city: City name\n"
" '''\n"
" async with httpx.AsyncClient() as c:\n"
" r = await c.get(\n"
" f'https://wttr.in/{city}'\n"
" '?format=j1',\n"
" timeout=10)\n"
" data = r.json()\n"
" current = data[\n"
" 'current_condition'][0]\n"
" return {\n"
" 'city': city,\n"
" 'temp_c': current['temp_C'],\n"
" 'condition': current[\n"
" 'weatherDesc'][0]['value'],\n"
" 'humidity': current[\n"
" 'humidity'],\n"
" }"
)
def get_oauth_middleware(self) -> str:
"""OAuth middleware for streamable HTTP."""
return (
"# auth.py - OAuth middleware\n"
"from jose import jwt\n"
"import os\n"
"\n"
"JWKS_URL = os.environ.get(\n"
" 'JWKS_URL')\n"
"AUDIENCE = os.environ.get(\n"
" 'AUDIENCE')\n"
"\n"
"async def auth_middleware(\n"
" request, call_next\n"
"):\n"
" '''Verify bearer token.'''\n"
" auth = request.headers.get(\n"
" 'Authorization', '')\n"
" if not auth.startswith(\n"
" 'Bearer '):\n"
" return Response(\n"
" status_code=401,\n"
" headers={\n"
" 'WWW-Authenticate':\n"
" 'Bearer '\n"
" 'resource_metadata='\n"
" f'{JWKS_URL}'\n"
" }\n"
" )\n"
" token = auth[7:]\n"
" try:\n"
" payload = jwt.decode(\n"
" token, JWKS_URL,\n"
" audience=AUDIENCE)\n"
" request.state.user = payload\n"
" except Exception:\n"
" return Response(\n"
" status_code=401)\n"
" return await call_next(request)"
)
def get_dockerfile(self) -> str:
"""Dockerfile for MCP API server."""
return (
"FROM python:3.12-slim\n"
"\n"
"WORKDIR /app\n"
"\n"
"COPY pyproject.toml .\n"
"RUN pip install -e .\n"
"\n"
"COPY src/ ./src/\n"
"\n"
"ENV MCP_TRANSPORT=stdio\n"
"\n"
"CMD [\"python\", \"-m\", "
"\"my_mcp_server\"]"
)
def get_production_checklist(self) -> list:
"""Production checklist for MCP API servers."""
return [
"Every tool has a clear, specific docstring",
"All API keys in environment variables, "
"never hardcoded",
"HTTP client has timeouts set "
"(httpx timeout=10-15)",
"Error responses return helpful messages, "
"not stack traces",
"Tools return JSON strings for "
"structured data",
"Dockerfile builds and runs cleanly",
"pyproject.toml has correct entry point",
"Tested with Claude Desktop or mcp dev",
"Type hints on all parameters for "
"auto-validation",
"TTL caching for frequent API calls",
"One tool = one job (focused tools)",
"OAuth middleware for remote deployment",
"WWW-Authenticate header for OAuth "
"discovery",
"Separate files: server, client, auth",
"Connection pooling via httpx.AsyncClient",
"Rate limiting awareness (handle 429)",
]
MCP API Server Checklist
- [ ] Use FastMCP with httpx for async HTTP calls to external REST APIs
- [ ] FastMCP handles JSON-RPC framing, schema generation from type hints, transport plumbing
- [ ] Three concerns, three files: server (MCP surface), client (REST API wrapper), auth (token verification)
- [ ] API client class: httpx.AsyncClient with base_url, auth headers, timeout=10.0
- [ ] API client methods: get, post, patch, delete with raise_for_status
- [ ] Tools call client methods and return JSON strings for structured data
- [ ] @mcp.tool() decorator auto-generates schema from type hints and docstrings
- [ ] One tool = one job — focused tools are easier for AI to decide when to call
- [ ] Type all parameters — MCP uses type hints for validation
- [ ] Write detailed docstrings — AI reads them to decide tool calls
- [ ] Handle errors — return useful error messages, not stack traces
- [ ] Add TTL caching — reduce API calls for frequent requests
- [ ] Set HTTP timeouts — httpx timeout=10-15 seconds
- [ ] All API keys in environment variables, never hardcoded
- [ ] Tools are like POST endpoints — they do something and may have side effects
- [ ] LLM calls tools when it needs to act
- [ ] Slack MCP: get_messages, send_message, list_channels using slack-sdk
- [ ] Notion MCP: search_pages, create_page, get_page using notion-client
- [ ] Jira MCP: search_issues, create_issue, update_issue via REST API v3 with httpx
- [ ] Weather MCP: get_weather using httpx to wttr.in or OpenWeatherMap
- [ ] Same structure works for Salesforce, internal APIs, proprietary databases
- [ ] stdio transport for local development (Claude Desktop, Cursor)
- [ ] streamable-http transport for remote production deployment
- [ ] streamable-http works with load balancers, gateways, TLS termination
- [ ] streamable-http is the only transport with a real auth model
- [ ] OAuth middleware for streamable-http: verify bearer token, return 401 with WWW-Authenticate
- [ ] WWW-Authenticate header is the discovery hook for OAuth
- [ ] Use python-jose for JWT verification
- [ ] Separate auth file for token verification logic
- [ ] FastMCP 1.0 incorporated into official MCP Python SDK in 2024
- [ ] FastMCP downloaded a million times a day, powers 70% of MCP servers
- [ ] MCP ecosystem has 17,000+ servers but most undocumented and unmaintained
- [ ] High-demand niches: database query tools, API integrations, data analysis, DevOps
- [ ] MCP prompts: reusable templates (e.g., sprint_review for Jira, summarize_items)
- [ ] MCP resources: static data (e.g., API documentation, configuration)
- [ ] Dockerfile for production deployment
- [ ] pyproject.toml with correct entry point: project.scripts
- [ ] Test with Claude Desktop or mcp dev (MCP Inspector) before deploying
- [ ] Handle rate limiting (HTTP 429) with retry and backoff
- [ ] Connection pooling via httpx.AsyncClient for performance
- [ ] Read MCP developer guide for protocol
- [ ] Read build MCP server for server creation
- [ ] Read MCP tools for tool development
- [ ] Read MCP database server for database servers
- [ ] Read MCP security for security risks
- [ ] Read MCP authentication for auth
- [ ] Test: tools call API correctly and return expected results
- [ ] Test: error handling returns helpful messages for API failures
- [ ] Test: timeouts prevent hanging on slow APIs
- [ ] Test: caching reduces redundant API calls
- [ ] Test: OAuth middleware rejects invalid tokens
- [ ] Test: Dockerfile builds and runs cleanly
- [ ] Document API endpoints, tools, auth method, and deployment process
FAQ
How do you build an MCP server that wraps a REST API?
Use FastMCP with httpx for async HTTP calls. Create an API client class, register tools that call the client, and run with stdio or streamable-http transport. WorkOS: "Practical guide to building an MCP server that wraps an existing REST API. Use official Python SDK with FastMCP. FastMCP handles JSON-RPC framing, schema generation from type hints, and transport plumbing. Three concerns, three files: server file holds MCP surface, client file wraps REST API, auth file verifies bearer tokens." MCP Template: "Production-ready template. api_client.py wraps any REST API. server.py defines MCP tools. Pattern: @mcp.tool() wrapping client.get/post calls." DEV Community: "Replace toy examples with actual API calls. Use httpx.AsyncClient for HTTP. Type parameters, write docstrings, add caching, handle errors, keep tools focused." Steps: (1) Create API client class with httpx. (2) Register @mcp.tool() functions calling client. (3) Set env vars for API URL and key. (4) Run with stdio for local, streamable-http for remote.
How do you integrate Slack, Notion, and Jira with MCP?
Build separate MCP servers for each service using their SDKs and APIs. Each server exposes tools for reading and writing to that service. Medium: "Step-by-step tutorial: Slack MCP server reads messages with slack-sdk. Notion MCP server searches pages with notion-client. Jira MCP server manages tickets via REST API. Tools are like POST endpoints — they do something and may have side effects. LLM calls them when it needs to act." Slack tools: get_messages, send_message, list_channels. Notion tools: search_pages, create_page, get_page. Jira tools: search_issues, create_issue, update_issue. Each server uses its respective SDK: slack-sdk, notion-client, Jira REST API via httpx. Deploy as HTTP servers for remote access. Same structure works for Salesforce, internal APIs, proprietary databases.
What is the API client pattern for MCP servers?
Separate API client logic from MCP tool definitions. Create a client class wrapping httpx, then register tools that call the client. WorkOS: "Three concerns, three files: server file holds MCP surface, client file wraps REST API, auth file verifies bearer tokens. Mixing these together gets ugly fast in any server with more than two or three tools. PantryClient class with httpx.AsyncClient, base_url, Authorization header, timeout=10.0. Methods: get, post, patch, aclose." MCP Template: "api_client.py: BASE_URL from env, API_KEY from env. client.get/post with params. Tools return JSON strings for structured data." Pattern: (1) API client class with httpx.AsyncClient. (2) base_url and auth headers from env vars. (3) Methods: get, post, patch, delete. (4) Tools call client methods and return JSON. (5) Error handling: raise_for_status, return error dict. (6) Timeout on all calls.
How do you handle authentication for MCP API servers?
Use OAuth middleware for streamable-http transport, environment variables for stdio, and bearer token verification. WorkOS: "FastMCP supports custom HTTP middleware on streamable HTTP transport. Add middleware that pulls token, verifies it, and on failure returns 401 with WWW-Authenticate header for discovery. streamable-http is the current standard for remote deployment. Works with load balancers, gateways, TLS termination. Only transport with a real auth model." MCP Template: "All API keys in environment variables, never hardcoded. HTTP client has timeouts set." Auth: (1) stdio: env vars for API keys. (2) HTTP: OAuth middleware with bearer token verification. (3) WWW-Authenticate header for OAuth discovery. (4) Separate auth file for token verification. (5) Use python-jose for JWT verification.
What are best practices for MCP API server production?
Use clear docstrings, env vars for secrets, HTTP timeouts, error handling, JSON string returns, Docker deployment, and test with MCP Inspector. MCP Template: "Production checklist: every tool has clear docstring, all API keys in env vars, HTTP client has timeouts (httpx.Client timeout=15), error responses return helpful messages not stack traces, tools return JSON strings, Dockerfile builds cleanly, pyproject.toml has correct entry point, tested with Claude Desktop or mcp dev." DEV Community: "Type your parameters — MCP uses them for validation. Write docstrings — they become tool description AI reads. Add caching — reduce API calls with TTL cache. Handle errors — return useful error messages, do not crash. Keep tools focused — one tool = one job." FastMCP: "FastMCP 1.0 incorporated into official MCP Python SDK in 2024. Downloaded a million times a day. Powers 70% of MCP servers across all languages. Best practices are built in." Best practices: (1) One tool = one job. (2) Type hints for validation. (3) Detailed docstrings. (4) TTL caching. (5) Error handling. (6) Timeouts. (7) Env vars for secrets. (8) Docker for deployment. (9) Test with MCP Inspector.
Want a self-hosted AI company brain that does all of this out of the box?
Book a demo →