This page covers creating your own MCP servers. See also: MCP Overview, Available Servers.
Building Your Own MCP Server
Prerequisites
| TypeScript | Python | |
|---|---|---|
| Runtime | Node.js 16+ | Python 3.10+ |
| Package manager | npm or pnpm | uv (recommended) or pip |
| SDK | @modelcontextprotocol/sdk | mcp[cli] |
| Schema | zod@3 | Built-in (type hints + docstrings) |
Choose Your Language
| Factor | TypeScript | Python |
|---|---|---|
| Schema definition | Explicit Zod objects | Auto from type hints + docstrings |
| Registration style | server.registerTool() | @mcp.tool() decorator |
| Scaffolding | npx @modelcontextprotocol/create-server | uv init + manual |
| Best for | Production web services | Rapid prototyping, data/ML pipelines |
Step 1: Create Project
# Option 1: Scaffold automatically
npx @modelcontextprotocol/create-server my-mcp-server
cd my-mcp-server && npm install
# Option 2: Manual
mkdir my-mcp-server && cd my-mcp-server
npm init -y
npm install @modelcontextprotocol/sdk zod@3
npm install -D @types/node typescript
Required package.json fields:
{
"type": "module",
"bin": { "my-mcp-server": "./build/index.js" },
"scripts": { "build": "tsc && chmod 755 build/index.js" }
}
uv init my-mcp-server
cd my-mcp-server
uv venv && source .venv/bin/activate
uv add "mcp[cli]" httpx
Step 2: Implement Server Core
// src/index.ts
import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
import { z } from "zod";
const server = new McpServer({
name: "my-server",
version: "1.0.0",
});
// Register tools here
async function main() {
const transport = new StdioServerTransport();
await server.connect(transport);
console.error("MCP Server running on stdio");
}
main().catch((error) => {
console.error("Fatal error:", error);
process.exit(1);
});
# server.py
from mcp.server.fastmcp import FastMCP
mcp = FastMCP("my-server")
# Register tools here
if __name__ == "__main__":
mcp.run(transport="stdio")
Step 3: Define Tools
Tools are the most important primitive. Aim for 5-20 tools per server.
// Basic tool
server.registerTool("greet", {
description: "Greet someone by name",
inputSchema: {
name: z.string().describe("Person's name"),
},
}, async ({ name }) => ({
content: [{ type: "text", text: `Hello, ${name}!` }],
}));
// Tool with structured output
server.registerTool("add", {
description: "Add two numbers",
inputSchema: { a: z.number(), b: z.number() },
outputSchema: z.object({ result: z.number() }),
}, async ({ a, b }) => ({
content: [{ type: "text", text: JSON.stringify({ result: a + b }) }],
structuredContent: { result: a + b },
}));
// Tool with annotations
server.registerTool("search", {
description: "Search the database",
inputSchema: { query: z.string() },
annotations: {
readOnlyHint: true,
destructiveHint: false,
idempotentHint: true,
},
}, async ({ query }) => ({
content: [{ type: "text", text: `Results for: ${query}` }],
}));
# Basic tool
@mcp.tool()
async def greet(name: str) -> str:
'''Greet someone by name.
Args:
name: Person's name
'''
return f"Hello, {name}!"
# Tool with structured output
from pydantic import BaseModel, Field
class AddResult(BaseModel):
result: float = Field(description="Sum of the two numbers")
@mcp.tool()
async def add(a: float, b: float) -> AddResult:
'''Add two numbers.
Args:
a: First number
b: Second number
'''
return AddResult(result=a + b)
Step 4: Add Resources and Prompts
// Static resource
server.registerResource(
"config", "config://app",
{ mimeType: "text/plain" },
async (uri) => ({
contents: [{ uri: uri.href, text: "key=value
env=production" }],
})
);
// Prompt template
server.registerPrompt("review-code", {
title: "Code Review",
argsSchema: z.object({
code: z.string(),
language: z.string().optional(),
}),
}, ({ code, language }) => ({
messages: [{
role: "user",
content: { type: "text", text: `Review this ${language || ""} code:
${code}` },
}],
}));
@mcp.resource("config://app")
def get_config() -> str:
'''Application configuration.'''
return "key=value
env=production"
@mcp.prompt()
def review_code(code: str, language: str = "python") -> str:
'''Review code for quality issues.'''
return f"Review this {language} code:
{code}"
Testing with MCP Inspector
# TypeScript server
npm run build
npx @modelcontextprotocol/inspector node build/index.js
# Python server
npx @modelcontextprotocol/inspector uv --directory /path/to/project run server.py
The Inspector provides: connection pane, tools tab (list and test), resources tab, prompts tab, and notifications pane.
Advanced Patterns
Database Server with Connection Pool
from contextlib import asynccontextmanager
from mcp.server.fastmcp import FastMCP
import asyncpg
@asynccontextmanager
async def app_lifespan(server):
pool = await asyncpg.create_pool(
"postgresql://user:pass@localhost/mydb",
min_size=2, max_size=10
)
try:
yield {"db_pool": pool}
finally:
await pool.close()
mcp = FastMCP("postgres-server", lifespan=app_lifespan)
@mcp.tool()
async def query_database(sql: str, ctx) -> str:
'''Execute a read-only SQL query.'''
if not sql.strip().upper().startswith("SELECT"):
return "Error: Only SELECT queries allowed"
pool = ctx.request_context.lifespan_context["db_pool"]
async with pool.acquire() as conn:
rows = await conn.fetch(sql)
return str([dict(r) for r in rows[:100]])
REST API Wrapper
import httpx
from mcp.server.fastmcp import FastMCP
mcp = FastMCP("github-api")
@mcp.tool()
async def search_repos(query: str, sort: str = "stars", limit: int = 10) -> str:
'''Search GitHub repositories.'''
async with httpx.AsyncClient() as client:
response = await client.get(
"https://api.github.com/search/repositories",
params={"q": query, "sort": sort, "per_page": min(limit, 30)},
headers={"Accept": "application/vnd.github.v3+json"},
)
response.raise_for_status()
return str([{
"name": r["full_name"],
"stars": r["stargazers_count"],
} for r in response.json()["items"]])
File System Server with Sandboxing
from pathlib import Path
from mcp.server.fastmcp import FastMCP
mcp = FastMCP("filesystem")
ALLOWED_DIRS = [Path("/Users/me/projects")]
def validate_path(file_path: str) -> Path:
resolved = Path(file_path).resolve()
for allowed in ALLOWED_DIRS:
if str(resolved).startswith(str(allowed.resolve())):
return resolved
raise ValueError(f"Access denied: {file_path}")
@mcp.tool()
def read_file(path: str) -> str:
'''Read file contents.'''
validated = validate_path(path)
return validated.read_text(encoding="utf-8")
Security & Production Checklist
| Category | Requirements |
|---|---|
| Security | Validate all inputs, access controls, rate limiting, sanitize outputs, HTTPS for auth, path traversal prevention, SQL injection prevention |
| Quality | 5-20 well-described tools, proper error handling (isError), progress reporting, structured logging, concise responses |
| Operations | Connection pooling, lifespan management (startup/shutdown), health monitoring, session management, graceful shutdown |
| Distribution | Clear README, absolute paths in examples, env var documentation, Inspector-tested, tested with Claude Desktop and Claude Code |
Packaging
| Platform | Publish | User Install |
|---|---|---|
| npm | npm publish --access public | npx -y @your-org/mcp-server-name |
| PyPI | python -m build && twine upload dist/* | uvx mcp-server-name |
| Docker | docker build -t mcp/server . | docker run -i --rm mcp/server |
Tool Design Guidelines
- 5-20 tools per server — too many overwhelms LLM tool selection
- Clear, descriptive names — the LLM reads them to decide which tool to use
- Comprehensive descriptions — primary way the LLM understands a tool
- Concise results — respect context window limits
- Use
isError: truefor business logic failures (not protocol errors)