4.2 — API Integration MCP — REST, GraphQL & WebSocket Servers
API Integration MCP — REST, GraphQL & WebSocket Servers
Every modern application talks to external APIs — payment gateways like JazzCash, product data from Daraz, weather for a Pakistan agriculture app, or a custom internal service your team built. Without MCP, you paste API responses into Claude and ask questions. With an API Integration MCP Server, Claude can call those APIs directly, chain multiple calls together, and take action — not just analyze. This lesson builds MCP servers for REST APIs, GraphQL endpoints, and real-time WebSocket connections.
Section 1: REST API MCP Server
REST APIs are the most common integration target. This pattern works for any REST API — JazzCash, Daraz Seller API, your own FastAPI backend, or third-party services.
Create rest-api-mcp-server.py:
#!/usr/bin/env python3
import httpx
import json
import os
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp import types
app = Server("rest-api-client")
# Base configuration — swap these for any REST API
API_BASE_URL = os.getenv("API_BASE_URL", "https://api.example.com")
API_KEY = os.getenv("API_KEY", "")
@app.list_tools()
async def list_tools():
return [
types.Tool(
name="api_get",
description="Make a GET request to a REST API endpoint",
inputSchema={
"type": "object",
"properties": {
"endpoint": {"type": "string", "description": "API endpoint path (e.g., /products/123)"},
"params": {"type": "object", "description": "Query parameters as key:value pairs"}
},
"required": ["endpoint"]
}
),
types.Tool(
name="api_post",
description="Make a POST request to a REST API endpoint",
inputSchema={
"type": "object",
"properties": {
"endpoint": {"type": "string"},
"body": {"type": "object", "description": "Request body as JSON"}
},
"required": ["endpoint", "body"]
}
),
types.Tool(
name="api_paginate",
description="Fetch multiple pages from a paginated REST API endpoint",
inputSchema={
"type": "object",
"properties": {
"endpoint": {"type": "string"},
"max_pages": {"type": "integer", "description": "Maximum pages to fetch (default 5)"}
},
"required": ["endpoint"]
}
)
]
@app.call_tool()
async def call_tool(name: str, arguments: dict):
headers = {"Authorization": f"Bearer {API_KEY}", "Content-Type": "application/json"}
async with httpx.AsyncClient(timeout=30) as client:
if name == "api_get":
url = f"{API_BASE_URL}{arguments['endpoint']}"
params = arguments.get("params", {})
response = await client.get(url, headers=headers, params=params)
return [types.TextContent(type="text", text=response.text)]
elif name == "api_post":
url = f"{API_BASE_URL}{arguments['endpoint']}"
response = await client.post(url, headers=headers, json=arguments["body"])
return [types.TextContent(type="text", text=response.text)]
elif name == "api_paginate":
url = f"{API_BASE_URL}{arguments['endpoint']}"
max_pages = arguments.get("max_pages", 5)
all_results = []
page = 1
while page <= max_pages:
response = await client.get(url, headers=headers, params={"page": page, "per_page": 100})
data = response.json()
if not data or (isinstance(data, list) and len(data) == 0):
break
all_results.extend(data if isinstance(data, list) else [data])
page += 1
return [types.TextContent(type="text", text=json.dumps(all_results, indent=2))]
Section 2: GraphQL MCP Server
GraphQL is common in modern SaaS platforms and some Pakistani startups using Hasura or Apollo. The key difference from REST is that you send queries as strings in the request body.
@app.list_tools()
async def list_tools():
return [
types.Tool(
name="graphql_query",
description="Execute a GraphQL query or mutation",
inputSchema={
"type": "object",
"properties": {
"query": {"type": "string", "description": "GraphQL query string"},
"variables": {"type": "object", "description": "GraphQL variables"}
},
"required": ["query"]
}
)
]
@app.call_tool()
async def call_tool(name: str, arguments: dict):
if name == "graphql_query":
GRAPHQL_ENDPOINT = os.getenv("GRAPHQL_URL")
async with httpx.AsyncClient() as client:
response = await client.post(
GRAPHQL_ENDPOINT,
json={"query": arguments["query"], "variables": arguments.get("variables", {})},
headers={"Authorization": f"Bearer {os.getenv('GRAPHQL_TOKEN')}"}
)
data = response.json()
if "errors" in data:
return [types.TextContent(type="text", text=f"GraphQL Error: {data['errors']}")]
return [types.TextContent(type="text", text=json.dumps(data["data"], indent=2))]
Section 3: WebSocket MCP Server
WebSocket integration is for real-time data — live trading prices, WhatsApp message streams, or IoT sensor data. The challenge is that MCP tools are request-response, not streaming. The pattern is to maintain a WebSocket connection in the background and let Claude poll for the latest data.
import asyncio
import websockets
import json
# Global buffer for latest WebSocket messages
ws_buffer = []
async def ws_listener(uri):
"""Background WebSocket listener — updates buffer"""
global ws_buffer
async with websockets.connect(uri) as websocket:
async for message in websocket:
ws_buffer.append(json.loads(message))
if len(ws_buffer) > 100: # Keep last 100 messages
ws_buffer = ws_buffer[-100:]
# Tool: get_latest_messages
@app.call_tool()
async def call_tool(name: str, arguments: dict):
if name == "get_ws_messages":
count = arguments.get("count", 10)
latest = ws_buffer[-count:] if ws_buffer else []
return [types.TextContent(type="text", text=json.dumps(latest, indent=2))]
Practice Lab
Exercise 1: Build the REST API MCP server and configure it to point to a free public API — try https://api.open-meteo.com (Pakistan weather data, no API key required). Connect it to Claude Code. Ask Claude: "What is the current weather forecast for Karachi for the next 3 days? Suggest if today is good for outdoor events." Verify Claude calls the API rather than guessing.
Exercise 2: Add an api_post tool connected to your own FastAPI backend (create a simple /echo endpoint that returns whatever you POST to it). Ask Claude to "POST a test message to the echo endpoint and confirm the response." This tests the full POST → response → analysis loop.
Exercise 3: Research one Pakistani API you could integrate for a real project: JazzCash payment API, Daraz Seller API, or Pakistan's TCS courier tracking API. Write a CLAUDE.md note describing: what the API does, what authentication it needs, and what 3 tools you would expose via MCP. You don't need to build it — this is the design exercise.
Key Takeaways
- REST API MCP Servers turn any HTTP API into a tool Claude can call directly — no more pasting API responses into chat windows
- The pagination tool is critical for production use — most real-world APIs return data in pages, and Claude needs to be able to fetch all of it for comprehensive analysis
- GraphQL MCP is essentially the same as REST MCP but with a single endpoint and query strings as the parameter — Claude is actually better at writing GraphQL queries than REST URLs since it can reason about the schema
- WebSocket MCP requires a buffer pattern — maintain the connection in the background and expose polling tools, not streaming tools, to Claude
Lesson Summary
Quiz: API Integration MCP — REST, GraphQL & WebSocket Servers
4 questions to test your understanding. Score 60% or higher to pass.