This page covers creating your own MCP servers. See also: MCP Overview, Available Servers.

Building Your Own MCP Server

Prerequisites

TypeScriptPython
RuntimeNode.js 16+Python 3.10+
Package managernpm or pnpmuv (recommended) or pip
SDK@modelcontextprotocol/sdkmcp[cli]
Schemazod@3Built-in (type hints + docstrings)

Choose Your Language

FactorTypeScriptPython
Schema definitionExplicit Zod objectsAuto from type hints + docstrings
Registration styleserver.registerTool()@mcp.tool() decorator
Scaffoldingnpx @modelcontextprotocol/create-serveruv init + manual
Best forProduction web servicesRapid 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

CategoryRequirements
SecurityValidate all inputs, access controls, rate limiting, sanitize outputs, HTTPS for auth, path traversal prevention, SQL injection prevention
Quality5-20 well-described tools, proper error handling (isError), progress reporting, structured logging, concise responses
OperationsConnection pooling, lifespan management (startup/shutdown), health monitoring, session management, graceful shutdown
DistributionClear README, absolute paths in examples, env var documentation, Inspector-tested, tested with Claude Desktop and Claude Code

Packaging

PlatformPublishUser Install
npmnpm publish --access publicnpx -y @your-org/mcp-server-name
PyPIpython -m build && twine upload dist/*uvx mcp-server-name
Dockerdocker build -t mcp/server .docker run -i --rm mcp/server

Tool Design Guidelines

  1. 5-20 tools per server — too many overwhelms LLM tool selection
  2. Clear, descriptive names — the LLM reads them to decide which tool to use
  3. Comprehensive descriptions — primary way the LLM understands a tool
  4. Concise results — respect context window limits
  5. Use isError: true for business logic failures (not protocol errors)