How to Build an MCP Server: Step-by-Step Guide

Model Context Protocol (MCP) servers are services that expose tools, resources, and prompts in a standard way so AI agents can call them safely and consistently. Instead of building one-off integrations for every agent × data source combination, MCP provides a shared interface that keeps auth, data formatting, and error handling predictable across systems.

This guide demonstrates how to build an MCP server step by step, including how to define capabilities, handle errors, and apply basic security patterns.

Why Build a Custom MCP Server?

Build a custom MCP server when pre-built servers cannot access your internal systems. 

Internal data warehouses are the most common reason. Custom schemas, organization-specific query patterns, and strict access controls rarely work with generic MCP servers. If your data lives behind internal auth or requires custom logic, you need to build.

Security and compliance are the second driver. Teams in regulated environments often require on-prem deployment, full audit trails, and complete visibility into how agents access data. Pre-built servers cannot meet these constraints.

Custom business logic is the final case. If your agent encodes multi-step workflows that reflect how your company operates, building makes sense. Use the “buy to learn, build to differentiate” rule: rely on pre-built servers for standard tools like GitHub or PostgreSQL, and build custom servers only for systems unique to your business.

What Do You Need Before Building?

Before building an MCP server, you need the following:

• Python 3.10 or higher: Required because the MCP SDK depends on modern Python language features. The official documentation recommends using uv for project management, though pip works for simpler setups.
  ‍
  
• MCP Python SDK: Install with pip install "mcp[cli]" or uv add "mcp[cli]". The official SDK includes FastMCP, a high-level framework that handles protocol details automatically. Import it as from mcp.server.fastmcp import FastMCP. Use the lower-level mcp.server APIs only if you need direct protocol control.
  
  
• A testing host for LLM interaction: Use Claude for Desktop as your primary MCP host to validate behavior before production. For local development and debugging, run the MCP Inspector via uv run mcp dev server.py to inspect tools, resources, prompts, and server behavior without configuring a full client.

Once your environment is set up, define what your MCP server exposes before writing code. Use @mcp.tool() for executable actions, @mcp.resource() for data access, and @mcp.prompt() for reusable templates. Each capability must declare clear type annotations and schemas so agents can understand how to use it.

Match capabilities to MCP primitives consistently. Tools execute logic, resources expose data, and prompts define structured templates. All schemas are registered at initialization, so missing or vague types will break agent interaction.

Choose transport based on deployment context. Local development and CLI-based usage rely on STDIO transport, where stdout is reserved strictly for JSON-RPC messages, and all logs must go to stderr. Production deployments serving multiple users should use Streamable HTTP transport, which avoids STDIO limitations and scales more reliably. Streamable HTTP is the recommended transport for remote servers as of protocol version 2025-06-18.

How to Build an MCP Server Step by Step

Here’s a walkthrough of building an MCP server, from project setup to exposing tools and resources.

1. Set Up Your Project Structure

Python (recommended approach using uv):

mkdir my-mcp-server
cd my-mcp-server
uv init
uv add "mcp[cli]"

Alternatively, install with pip:

pip install "mcp[cli]"

TypeScript:

mkdir my-mcp-server
cd my-mcp-server
npm init -y
npm install @modelcontextprotocol/sdk zod
npm install -D @types/node typescript

Create a tsconfig.json file:

{
  "compilerOptions": {
    "target": "ES2022",
    "module": "Node16",
    "moduleResolution": "Node16",
    "outDir": "./build",
    "rootDir": "./src",
    "strict": true,
    "esModuleInterop": true,
    "skipLibCheck": true,
    "forceConsistentCasingInFileNames": true
  },
  "include": ["src/**/*"],
  "exclude": ["node_modules"]
}

Update package.json to add the module type and build script:

{
  "type": "module",
  "scripts": {
    "build": "tsc && chmod 755 build/index.js"
  }
}

2. Create the Server Instance

Python with FastMCP (recommended):

FastMCP is included in the official MCP SDK and provides the simplest path to a working server. Decorators handle schema generation, parameter validation, and protocol compliance.

import json
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("my-mcp-server")

# Placeholder - replace with your actual database logic
def execute_query(query: str) -> list:
    return [{"id": 1, "result": "sample data"}]

# Placeholder - replace with your actual file reading logic
def read_file(path: str) -> str:
    return f"Contents of {path}"

@mcp.tool()
def query_database(query: str) -> str:
    """Execute SQL queries against the database."""
    results = execute_query(query)
    return json.dumps(results, indent=2)

@mcp.resource("file:///logs/app.log")
def get_logs() -> str:
    """Application logs."""
    return read_file("/var/logs/app.log")
```

---

TypeScript with the low-level SDK:

```typescript
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-mcp-server",
  version: "1.0.0",
});

// Placeholder - replace with your actual database logic
async function executeQuery(query: string): Promise<object[]> {
  return [{ id: 1, result: "sample data" }];
}

// Define and register a tool
server.tool(
  "query_database",
  { query: z.string().describe("SQL query to execute") },
  async ({ query }) => {
    const results = await executeQuery(query);
    return {
      content: [{ type: "text", text: JSON.stringify(results, null, 2) }],
    };
  }
);
```

---

// Connect via stdio transport
const transport = new StdioServerTransport();
await server.connect(transport);
console.error("MCP Server running on stdio");

3. Define Your Tools

Tools let agents execute actions. Each tool requires a name, description, and input schema. The SDK validates inputs against JSON Schema Draft 2020-12.

Python example with error handling:

import json
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("database-server")

# Placeholder - replace with your actual database logic
def execute_query(query: str) -> list:
    return [{"id": 1, "result": "sample data"}]

@mcp.tool()
def query_database(query: str) -> str:
    ...
```

---

    """Execute SQL queries against the database.
    
    Args:
        query: SQL query to execute
    """
    if not query or not query.strip():
        return "Error: 'query' parameter is required. Please provide a valid SQL query."
    
    try:
        results = execute_query(query)
        return f"Query executed successfully. Results:\n{json.dumps(results, indent=2)}"
    except Exception as e:
        return f"Query execution failed: {str(e)}"

TypeScript example with error handling:

// Placeholder - replace with your actual database logic
async function executeQuery(query: string): Promise<object[]> {
  return [{ id: 1, result: "sample data" }];
}
```
server.tool(
  "query_database",
  { 
    query: z.string().min(1).describe("SQL query to execute") 
  },
  async ({ query }) => {
    try {
      const results = await executeQuery(query);
      return {
        content: [{
          type: "text",
          text: `Query executed successfully. Results:\n${JSON.stringify(results, null, 2)}`
        }]
      };
    } catch (error) {
      return {
        content: [{
          type: "text",
          text: `Query failed: ${error instanceof Error ? error.message : String(error)}. Check your SQL syntax and try again.`
        }],
        isError: true
      };
    }
  }
);

Return errors as tool results, not protocol errors. The agent needs to understand what went wrong and potentially retry with different parameters.

4. Add Resources for Context

Resources provide read-only data access. They use URIs to identify different data sources and return content in formats agents can understand.

Python:

# Placeholder - replace with your actual file reading logic
def read_file(path: str) -> str:
    return f"Contents of {path}"
```

@mcp.resource("config://settings")
def get_settings() -> str:
    """Application configuration settings."""
    return json.dumps({
        "theme": "dark",
        "language": "en",
        "debug": False
    })

@mcp.resource("file://documents/{name}")
def read_document(name: str) -> str:
    """Read a document by name."""
    return read_file(f"/documents/{name}")

TypeScript:

server.resource(
  "config://settings",
  "Application Settings",
  async () => ({
    contents: [{
      uri: "config://settings",
      mimeType: "application/json",
      text: JSON.stringify({
        theme: "dark",
        language: "en",
        debug: false
      })
    }]
  })
);

5. Create Prompts for Common Workflows

Prompts are reusable, parameterized instruction templates that servers expose to standardize how AI agents interact with them for recurring tasks.

Python:

@mcp.prompt()
def code_review(code: str, language: str = "python") -> str:
    """Generate a code review prompt."""
    return f"Please review this {language} code:\n\n{code}"

TypeScript:

server.prompt(
  "code_review",
  { 
    code: z.string().describe("Code to review"),
    language: z.string().optional().describe("Programming language")
  },
  async ({ code, language = "python" }) => ({
    messages: [{
      role: "user",
      content: {
        type: "text",
        text: `Please review this ${language} code:\n\n${code}`
      }
    }]
  })
);

6. Configure Transport and Run

Choose stdio for local development or Streamable HTTP for remote deployments. Streamable HTTP is the recommended transport for production as of protocol version 2025-06-18.

Python with stdio (local development):

from mcp.server.fastmcp import FastMCP

mcp = FastMCP("my-server")

@mcp.tool()
def hello(name: str = "World") -> str:
    """Say hello to someone."""
    return f"Hello, {name}!"

if __name__ == "__main__":
    mcp.run(transport="stdio")

Python with Streamable HTTP (production):

if __name__ == "__main__":
    mcp.run(transport="streamable-http")

TypeScript with stdio:

import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";

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);
});

Run your server:

# Python
uv run server.py
# or
python server.py

# TypeScript
npm run build
node build/index.js

How to Test and Debug MCP Servers

Use MCP Inspector

The MCP Inspector provides a browser-based testing environment with no installation required. It runs directly through npx.

For Node.js servers:

npx @modelcontextprotocol/inspector node path/to/server/index.js

For Python servers:

npx @modelcontextprotocol/inspector python path/to/server.py

‍For PyPI-packaged Python servers:

npx @modelcontextprotocol/inspector uvx mcp-server-git --repository ~/code

For npm-packaged servers:

npx -y @modelcontextprotocol/inspector npx @modelcontextprotocol/server-filesystem ~/Desktop

The Inspector UI opens at http://localhost:6274 in your browser. It provides debugging capabilities including resource inspection, prompt testing, tool execution, and real-time notification monitoring.

Test resources, prompts, and tool invocations directly through the Inspector UI before connecting to Claude Desktop. Verify that error responses include actionable guidance for agents and that resources display correct MIME types.

Connect to Claude Desktop

Configure Claude Desktop to test your server with a real AI client. Edit the configuration file located at:

macOS: ~/Library/Application Support/Claude/claude_desktop_config.json
Windows: %APPDATA%\Claude\claude_desktop_config.json

Use this JSON structure to add your server:

{
  "mcpServers": {
    "my-server": {
      "command": "node",
      "args": ["/absolute/path/to/dist/index.js"],
      "env": {
        "API_KEY": "your-api-key"
      }
    }
  }
}

This configuration follows the format specified in the MCP documentation for Claude Desktop configuration files. The three required components are:

• command: The executable to run (e.g., "node", "npx", "python")
• args: Array of command-line arguments, with file paths specified as absolute paths
• env (optional): Environment variables to pass to the server process

Restart Claude Desktop completely after saving the configuration. Use the MCP Inspector tool (npx @modelcontextprotocol/inspector) to verify your server's tools appear correctly and respond to requests. In Claude Desktop, you can test the connection by interacting with your MCP server's exposed tools in conversations.

Handle Errors Properly

Return errors as tool results, not protocol exceptions. The agent needs to understand what went wrong and, if necessary, retry with different parameters. Log every request with unique identifiers for debugging.

import logging
import uuid
import asyncio
from mcp.server.fastmcp import FastMCP

logger = logging.getLogger(__name__)
mcp = FastMCP("my-server")

async def fetch_api(location: str) -> dict:
    """Example async API call."""
    await asyncio.sleep(0.1)
    return {"location": location, "status": "ok"}

@mcp.tool()
async def fetch_data(location: str) -> str:
    """Fetch data for a given location."""
    request_id = str(uuid.uuid4())
    logger.info(f"[{request_id}] Fetching data for: {location}")

    if not location:
        logger.warning(f"[{request_id}] Missing location parameter")
        return "Error: Location parameter is required"

    try:
        result = await fetch_api(location)
        logger.info(f"[{request_id}] Successfully fetched data for {location}")
        return str(result)
    except ValueError as e:
        logger.warning(f"[{request_id}] Invalid input: {e}")
        return f"Invalid input: {e}"
    except Exception:
        logger.exception(f"[{request_id}] Server error in fetch_data")
        return "Internal server error occurred"

This pattern provides several benefits:

• Request IDs let you trace issues through logs
• Distinguishing client errors (invalid input) from server errors (internal failures) helps with debugging
• Returning human-readable error messages allows agents to understand what went wrong and adjust their approach.

For TypeScript servers, follow the same pattern:

import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
import { z } from "zod";
import { v4 as uuidv4 } from "uuid";

const server = new McpServer({
  name: "my-server",
  version: "1.0.0",
});

server.tool(
  "fetch_data",
  { location: z.string().describe("The location to fetch data for") },
  async ({ location }) => {
    const requestId = uuidv4();
    console.error(`[${requestId}] Fetching data for: ${location}`);
    
    try {
      if (!location) {
        console.error(`[${requestId}] Client error: Missing location parameter`);
        return {
          isError: true,
          content: [{ type: "text", text: "Error: Location parameter is required" }]
        };
      }
      
      const result = await fetchApi(location);
      console.error(`[${requestId}] Successfully fetched data for ${location}`);
      return {
        content: [{ type: "text", text: String(result) }]
      };
    } catch (error) {
      console.error(`[${requestId}] Server error:`, error);
      return {
        isError: true,
        content: [{ type: "text", text: "Internal server error occurred" }]
      };
    }
  }
);

Note that TypeScript servers using STDIO transport must log to stderr (using console.error), not stdout. The STDIO transport reserves stdout exclusively for JSON-RPC messages.

How to Deploy an MCP Server to Production?

Once your MCP server runs locally, the next step is to deploy it in a way that’s reliable, secure, and easy to operate. Here’s how to do it:

Local Deployment via STDIO

STDIO transport works well for single-user, local deployments such as desktop integrations. The client launches your server as a child process and communicates through stdin/stdout. This offers the lowest latency compared to network-based transports and requires the simplest configuration.

Package your server as an npm package or Python module that users install locally. They configure their MCP client to launch your executable with appropriate environment variables.

from mcp.server.fastmcp import FastMCP

mcp = FastMCP("my-local-server")

@mcp.tool()
def process_data(input: str) -> str:
    """Process data locally."""
    return f"Processed: {input}"

if __name__ == "__main__":
    mcp.run()  # Uses STDIO transport by default

Remote Deployment via Streamable HTTP

Production deployments serving multiple clients need HTTP transport. Streamable HTTP is the recommended approach for remote deployments as of protocol version 2025-06-26. It provides better performance than the legacy SSE approach and simplifies infrastructure by using a single /mcp endpoint for all communication.

from mcp.server.fastmcp import FastMCP

mcp = FastMCP("my-remote-server")

@mcp.tool()
def fetch_data(query: str) -> str:
    """Fetch data from the server."""
    return f"Results for: {query}"

if __name__ == "__main__":
    mcp.run(
        transport="streamable-http",
        host="0.0.0.0",
        port=8000
    )

For stateless deployments that scale horizontally:

mcp = FastMCP("my-stateless-server", stateless_http=True)

if __name__ == "__main__":
    mcp.run(transport="streamable-http", host="0.0.0.0", port=8000)

Deploy your server behind a reverse proxy (nginx, Cloudflare, AWS ALB) with TLS termination. Handle encryption at the infrastructure layer rather than in the application code. This approach simplifies certificate management and follows standard production practices.

Security Considerations

Authentication: Use OAuth 2.1 with PKCE for production authentication. The MCP specification defines servers as OAuth Resource Servers that validate tokens issued by separate Authorization Servers. For simpler setups, API keys stored in environment variables or secrets managers provide basic protection.

import os
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("secure-server")

def get_api_key() -> str:
    """Load API key from environment."""
    key = os.getenv("MCP_API_KEY")
    if not key:
        raise ValueError("MCP_API_KEY environment variable not set")
    return key

@mcp.tool()
def authenticated_action(request: str) -> str:
    """Perform an action that requires authentication."""
    api_key = get_api_key()
    # Use api_key for downstream service calls
    return f"Authenticated request processed: {request}"

Input validation: Validate all inputs to prevent injection attacks. Sanitize strings, check numeric ranges, and verify enum values match expected options.

import re
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("validated-server")

MAX_INPUT_LENGTH = 10000

def sanitize_string(input_str: str) -> str:
    """Sanitize string input."""
    if len(input_str) > MAX_INPUT_LENGTH:
        raise ValueError(f"Input exceeds maximum length of {MAX_INPUT_LENGTH}")
    
    # Remove control characters
    cleaned = re.sub(r'[\x00-\x1F\x7F]', '', input_str)
    return cleaned

def validate_numeric_range(value: float, min_val: float, max_val: float) -> float:
    """Validate numeric value is within acceptable range."""
    if not (min_val <= value <= max_val):
        raise ValueError(f"Value {value} out of range [{min_val}, {max_val}]")
    return value

@mcp.tool()
def process_user_input(text: str, count: int) -> str:
    """Process user input with validation."""
    clean_text = sanitize_string(text)
    valid_count = validate_numeric_range(count, 1, 1000)
    return f"Processed '{clean_text}' with count {valid_count}"

Least privilege file access: Restrict file system access to specific directories. Normalize paths to prevent traversal attacks.

import os
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("file-server")

ALLOWED_DIRECTORIES = ["/app/data", "/app/config"]

@mcp.tool()
def read_file(path: str) -> str:
    """Read file contents with path traversal protection."""
    # Normalize path to prevent traversal attacks
    normalized = os.path.normpath(os.path.abspath(path))
    
    # Validate path is within allowed directories
    if not any(normalized.startswith(d) for d in ALLOWED_DIRECTORIES):
        return f"Access denied: Path '{path}' is outside allowed directories"
    
    try:
        with open(normalized, 'r') as f:
            return f.read()
    except FileNotFoundError:
        return f"File not found: {path}"
    except Exception as e:
        return f"Error reading file: {str(e)}"

Environment variable management: Never include credentials in code or configuration files committed to version control.

import os
from dataclasses import dataclass

@dataclass
class ServerConfig:
    """Server configuration loaded from environment."""
    api_key: str
    database_url: str
    debug: bool
    
    @classmethod
    def from_environment(cls) -> "ServerConfig":
        api_key = os.getenv("MCP_API_KEY")
        database_url = os.getenv("DATABASE_URL")
        
        if not api_key:
            raise ValueError("MCP_API_KEY environment variable required")
        if not database_url:
            raise ValueError("DATABASE_URL environment variable required")
        
        return cls(
            api_key=api_key,
            database_url=database_url,
            debug=os.getenv("DEBUG", "false").lower() == "true"
        )

For comprehensive security guidance, refer to the MCP security documentation and the authorization specification.

What Are Common Mistakes When Building MCP Servers?

The table below shows the most common mistakes teams make when building MCP servers and how to avoid them.

When Should You Use Pre-Built Infrastructure Instead?

Build a custom MCP server only when you need to expose proprietary systems with no existing alternative. Internal data warehouses, company-specific business logic, and strict compliance requirements can justify custom development.

For most other cases, pre-built infrastructure is the better choice. Teams lose weeks maintaining brittle scripts that connect agents to tools like Notion, Slack, and Google Drive. APIs change, tokens expire, and integrations break, pulling engineers away from building agent behavior.

Airbyte’s Agent Engine addresses this directly. The PyAirbyte MCP Server lets teams manage data connectors through AI assistants like Claude Desktop, Cursor, and Cline, without writing custom integration code. Authentication, token refresh, and API changes are handled automatically across hundreds of connectors.

For customer-facing use cases, Airbyte’s embeddable widget allows end users to connect their own data through a white-label UI. Enterprise teams also get deployment flexibility, including on-prem and hybrid options, with built-in access controls and compliance support.

Join the private beta to see how Airbyte Embedded helps you move from integration work to production-ready agents fast.

Frequently Asked Questions

Do I need to know MCP specification details to build a server?

Use the official SDKs or FastMCP to handle protocol implementation. The SDKs manage JSON-RPC messaging, capability negotiation, and transport layers while you define tools, resources, and prompts through simple APIs.

Can one MCP server expose tools and resources together?

Yes, a single server can expose any combination of tools, resources, and prompts. During initialization with the MCP protocol, servers may advertise supported capabilities, but there is no formal three-step handshake and clients only implement request handlers for the capabilities they wish to use.

Which MCP clients work with custom servers?

Most clients implementing the MCP protocol, such as Cursor and Cline, can connect to custom servers through standard configuration, provided the server adheres to the protocol. However, Claude Desktop does not currently support connecting to remote MCP servers.

How do I update my server without breaking existing clients?

MCP uses date-based protocol versions with explicit negotiation during initialization. Clients send their supported version in the initialize request, and servers respond with a compatible version. The protocol maintains backward compatibility within versions.

What's the difference between FastMCP and the official MCP SDK?

FastMCP provides a higher-level abstraction layer built on the official Python SDK, reducing boilerplate through decorator patterns. The official SDK gives you maximum control over protocol implementation.