Model Context Protocol (MCP): Build Custom AI Tool Integrations

What is Model Context Protocol (MCP)?

Model Context Protocol (MCP) is an open standard developed by Anthropic that enables AI models to securely connect with external data sources and tools. Think of MCP as a universal adapter that lets LLMs interact with databases, APIs, file systems, and custom business logic.

Why MCP Matters

Traditional Approach	MCP Approach
Custom integrations per tool	Standardized protocol
Hardcoded API connections	Dynamic tool discovery
Limited context	Rich, real-time data
Manual data piping	Automatic context injection
Security concerns	Built-in sandboxing

MCP Architecture

┌─────────────────────────────────────────────────────────────┐
│                     MCP Host (Claude)                        │
│  ┌─────────────────────────────────────────────────────┐    │
│  │               MCP Client                             │    │
│  └───────────────────┬─────────────────────────────────┘    │
└──────────────────────┼──────────────────────────────────────┘
                       │ JSON-RPC over stdio/SSE
       ┌───────────────┼───────────────┐
       ▼               ▼               ▼
┌─────────────┐ ┌─────────────┐ ┌─────────────┐
│ MCP Server  │ │ MCP Server  │ │ MCP Server  │
│  (Database) │ │   (GitHub)  │ │  (Custom)   │
└─────────────┘ └─────────────┘ └─────────────┘
       │               │               │
       ▼               ▼               ▼
   PostgreSQL      GitHub API     Your Tools

Get the Code

Clone the working examples from GitHub:

git clone https://github.com/Moshiour027/techyowls-io-blog-public.git
cd techyowls-io-blog-public/model-context-protocol-mcp-guide

Core MCP Concepts

Resources

Resources expose data to the LLM. They’re like read-only endpoints that provide context.

# Resources provide context data
@server.list_resources()
async def list_resources():
    return [
        Resource(
            uri="file:///logs/app.log",
            name="Application Logs",
            description="Recent application log entries",
            mimeType="text/plain"
        )
    ]

Tools

Tools let the LLM take actions. They’re callable functions with defined inputs and outputs.

# Tools enable LLM actions
@server.list_tools()
async def list_tools():
    return [
        Tool(
            name="query_database",
            description="Execute SQL queries",
            inputSchema={
                "type": "object",
                "properties": {
                    "query": {"type": "string", "description": "SQL query"}
                },
                "required": ["query"]
            }
        )
    ]

Prompts

Prompts are reusable templates that guide LLM interactions.

# Prompts provide interaction templates
@server.list_prompts()
async def list_prompts():
    return [
        Prompt(
            name="analyze_data",
            description="Analyze dataset and provide insights",
            arguments=[
                PromptArgument(name="dataset", description="Dataset name", required=True)
            ]
        )
    ]

Building Your First MCP Server (Python)

Project Setup

# Create project directory
mkdir mcp-server-demo && cd mcp-server-demo

# Create virtual environment
python -m venv venv
source venv/bin/activate  # On Windows: venv\Scripts\activate

# Install MCP SDK
pip install mcp anthropic python-dotenv

# Project structure
mkdir -p src
touch src/__init__.py src/server.py

Basic MCP Server

# src/server.py
import asyncio
import json
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import (
    Resource,
    Tool,
    TextContent,
    CallToolResult
)

# Initialize server
server = Server("demo-server")

# In-memory data store
data_store = {
    "users": [
        {"id": 1, "name": "Alice", "email": "alice@example.com"},
        {"id": 2, "name": "Bob", "email": "bob@example.com"}
    ],
    "products": [
        {"id": 1, "name": "Widget", "price": 29.99},
        {"id": 2, "name": "Gadget", "price": 49.99}
    ]
}

@server.list_resources()
async def list_resources():
    """List available data resources."""
    return [
        Resource(
            uri="data://users",
            name="Users Database",
            description="List of all users in the system",
            mimeType="application/json"
        ),
        Resource(
            uri="data://products",
            name="Products Catalog",
            description="Available products and pricing",
            mimeType="application/json"
        )
    ]

@server.read_resource()
async def read_resource(uri: str):
    """Read resource data by URI."""
    if uri == "data://users":
        return json.dumps(data_store["users"], indent=2)
    elif uri == "data://products":
        return json.dumps(data_store["products"], indent=2)
    raise ValueError(f"Unknown resource: {uri}")

@server.list_tools()
async def list_tools():
    """List available tools."""
    return [
        Tool(
            name="add_user",
            description="Add a new user to the database",
            inputSchema={
                "type": "object",
                "properties": {
                    "name": {"type": "string", "description": "User's full name"},
                    "email": {"type": "string", "description": "User's email address"}
                },
                "required": ["name", "email"]
            }
        ),
        Tool(
            name="search_products",
            description="Search products by name",
            inputSchema={
                "type": "object",
                "properties": {
                    "query": {"type": "string", "description": "Search query"}
                },
                "required": ["query"]
            }
        ),
        Tool(
            name="calculate_total",
            description="Calculate total price for product IDs",
            inputSchema={
                "type": "object",
                "properties": {
                    "product_ids": {
                        "type": "array",
                        "items": {"type": "integer"},
                        "description": "List of product IDs"
                    }
                },
                "required": ["product_ids"]
            }
        )
    ]

@server.call_tool()
async def call_tool(name: str, arguments: dict) -> CallToolResult:
    """Execute tool with given arguments."""

    if name == "add_user":
        new_id = max(u["id"] for u in data_store["users"]) + 1
        new_user = {
            "id": new_id,
            "name": arguments["name"],
            "email": arguments["email"]
        }
        data_store["users"].append(new_user)
        return CallToolResult(
            content=[TextContent(
                type="text",
                text=f"Created user with ID {new_id}: {new_user['name']}"
            )]
        )

    elif name == "search_products":
        query = arguments["query"].lower()
        results = [
            p for p in data_store["products"]
            if query in p["name"].lower()
        ]
        return CallToolResult(
            content=[TextContent(
                type="text",
                text=json.dumps(results, indent=2)
            )]
        )

    elif name == "calculate_total":
        product_ids = arguments["product_ids"]
        total = sum(
            p["price"] for p in data_store["products"]
            if p["id"] in product_ids
        )
        return CallToolResult(
            content=[TextContent(
                type="text",
                text=f"Total: ${total:.2f}"
            )]
        )

    raise ValueError(f"Unknown tool: {name}")

async def main():
    """Run the MCP server."""
    async with stdio_server() as (read_stream, write_stream):
        await server.run(
            read_stream,
            write_stream,
            server.create_initialization_options()
        )

if __name__ == "__main__":
    asyncio.run(main())

Running the Server

# Run directly (for testing)
python src/server.py

# Or via module
python -m src.server

Building MCP Server with TypeScript

TypeScript Setup

# Create TypeScript project
mkdir mcp-ts-server && cd mcp-ts-server
npm init -y

# Install dependencies
npm install @modelcontextprotocol/sdk zod
npm install -D typescript @types/node ts-node

# Initialize TypeScript
npx tsc --init

TypeScript MCP Server

// src/server.ts
import { Server } from "@modelcontextprotocol/sdk/server/index.js";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
import {
  CallToolRequestSchema,
  ListResourcesRequestSchema,
  ListToolsRequestSchema,
  ReadResourceRequestSchema,
} from "@modelcontextprotocol/sdk/types.js";
import { z } from "zod";

// Create server instance
const server = new Server(
  {
    name: "typescript-mcp-server",
    version: "1.0.0",
  },
  {
    capabilities: {
      resources: {},
      tools: {},
    },
  }
);

// Data store
interface Task {
  id: number;
  title: string;
  completed: boolean;
  priority: "low" | "medium" | "high";
}

let tasks: Task[] = [
  { id: 1, title: "Learn MCP", completed: false, priority: "high" },
  { id: 2, title: "Build server", completed: false, priority: "medium" },
];

// List resources
server.setRequestHandler(ListResourcesRequestSchema, async () => {
  return {
    resources: [
      {
        uri: "tasks://all",
        name: "All Tasks",
        description: "List of all tasks in the system",
        mimeType: "application/json",
      },
      {
        uri: "tasks://pending",
        name: "Pending Tasks",
        description: "Tasks that are not completed",
        mimeType: "application/json",
      },
    ],
  };
});

// Read resources
server.setRequestHandler(ReadResourceRequestSchema, async (request) => {
  const uri = request.params.uri;

  if (uri === "tasks://all") {
    return {
      contents: [
        {
          uri,
          mimeType: "application/json",
          text: JSON.stringify(tasks, null, 2),
        },
      ],
    };
  }

  if (uri === "tasks://pending") {
    const pending = tasks.filter((t) => !t.completed);
    return {
      contents: [
        {
          uri,
          mimeType: "application/json",
          text: JSON.stringify(pending, null, 2),
        },
      ],
    };
  }

  throw new Error(`Unknown resource: ${uri}`);
});

// List tools
server.setRequestHandler(ListToolsRequestSchema, async () => {
  return {
    tools: [
      {
        name: "create_task",
        description: "Create a new task",
        inputSchema: {
          type: "object",
          properties: {
            title: { type: "string", description: "Task title" },
            priority: {
              type: "string",
              enum: ["low", "medium", "high"],
              description: "Task priority",
            },
          },
          required: ["title"],
        },
      },
      {
        name: "complete_task",
        description: "Mark a task as completed",
        inputSchema: {
          type: "object",
          properties: {
            id: { type: "number", description: "Task ID" },
          },
          required: ["id"],
        },
      },
      {
        name: "delete_task",
        description: "Delete a task by ID",
        inputSchema: {
          type: "object",
          properties: {
            id: { type: "number", description: "Task ID" },
          },
          required: ["id"],
        },
      },
    ],
  };
});

// Input schemas
const CreateTaskSchema = z.object({
  title: z.string(),
  priority: z.enum(["low", "medium", "high"]).default("medium"),
});

const TaskIdSchema = z.object({
  id: z.number(),
});

// Handle tool calls
server.setRequestHandler(CallToolRequestSchema, async (request) => {
  const { name, arguments: args } = request.params;

  if (name === "create_task") {
    const { title, priority } = CreateTaskSchema.parse(args);
    const newTask: Task = {
      id: Math.max(...tasks.map((t) => t.id), 0) + 1,
      title,
      completed: false,
      priority,
    };
    tasks.push(newTask);
    return {
      content: [
        {
          type: "text",
          text: `Created task #${newTask.id}: "${title}" (${priority} priority)`,
        },
      ],
    };
  }

  if (name === "complete_task") {
    const { id } = TaskIdSchema.parse(args);
    const task = tasks.find((t) => t.id === id);
    if (!task) {
      return {
        content: [{ type: "text", text: `Task #${id} not found` }],
        isError: true,
      };
    }
    task.completed = true;
    return {
      content: [{ type: "text", text: `Completed task #${id}: "${task.title}"` }],
    };
  }

  if (name === "delete_task") {
    const { id } = TaskIdSchema.parse(args);
    const index = tasks.findIndex((t) => t.id === id);
    if (index === -1) {
      return {
        content: [{ type: "text", text: `Task #${id} not found` }],
        isError: true,
      };
    }
    const [deleted] = tasks.splice(index, 1);
    return {
      content: [{ type: "text", text: `Deleted task #${id}: "${deleted.title}"` }],
    };
  }

  throw new Error(`Unknown tool: ${name}`);
});

// Start server
async function main() {
  const transport = new StdioServerTransport();
  await server.connect(transport);
  console.error("MCP TypeScript server running on stdio");
}

main().catch(console.error);

Package Configuration

{
  "name": "mcp-ts-server",
  "version": "1.0.0",
  "type": "module",
  "main": "dist/server.js",
  "scripts": {
    "build": "tsc",
    "start": "node dist/server.js",
    "dev": "ts-node --esm src/server.ts"
  }
}

Database Integration Example

PostgreSQL MCP Server

# src/postgres_server.py
import asyncio
import json
from typing import Any
import asyncpg
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import Resource, Tool, TextContent, CallToolResult

server = Server("postgres-mcp")

# Database connection pool
pool: asyncpg.Pool = None

async def init_db():
    global pool
    pool = await asyncpg.create_pool(
        host="localhost",
        port=5432,
        user="postgres",
        password="password",
        database="mydb",
        min_size=2,
        max_size=10
    )

@server.list_resources()
async def list_resources():
    """List database tables as resources."""
    async with pool.acquire() as conn:
        tables = await conn.fetch("""
            SELECT table_name
            FROM information_schema.tables
            WHERE table_schema = 'public'
        """)

    return [
        Resource(
            uri=f"postgres://table/{t['table_name']}",
            name=f"Table: {t['table_name']}",
            description=f"Data from {t['table_name']} table",
            mimeType="application/json"
        )
        for t in tables
    ]

@server.read_resource()
async def read_resource(uri: str):
    """Read table data."""
    if uri.startswith("postgres://table/"):
        table_name = uri.split("/")[-1]
        # Prevent SQL injection
        if not table_name.isidentifier():
            raise ValueError("Invalid table name")

        async with pool.acquire() as conn:
            rows = await conn.fetch(f"SELECT * FROM {table_name} LIMIT 100")
            data = [dict(row) for row in rows]

        return json.dumps(data, indent=2, default=str)

    raise ValueError(f"Unknown resource: {uri}")

@server.list_tools()
async def list_tools():
    return [
        Tool(
            name="query",
            description="Execute a SELECT query (read-only)",
            inputSchema={
                "type": "object",
                "properties": {
                    "sql": {
                        "type": "string",
                        "description": "SELECT SQL query"
                    }
                },
                "required": ["sql"]
            }
        ),
        Tool(
            name="describe_table",
            description="Get table schema information",
            inputSchema={
                "type": "object",
                "properties": {
                    "table_name": {
                        "type": "string",
                        "description": "Name of the table"
                    }
                },
                "required": ["table_name"]
            }
        )
    ]

@server.call_tool()
async def call_tool(name: str, arguments: dict) -> CallToolResult:
    if name == "query":
        sql = arguments["sql"].strip()
        # Only allow SELECT queries
        if not sql.upper().startswith("SELECT"):
            return CallToolResult(
                content=[TextContent(
                    type="text",
                    text="Error: Only SELECT queries are allowed"
                )],
                isError=True
            )

        async with pool.acquire() as conn:
            rows = await conn.fetch(sql)
            data = [dict(row) for row in rows]

        return CallToolResult(
            content=[TextContent(
                type="text",
                text=json.dumps(data, indent=2, default=str)
            )]
        )

    elif name == "describe_table":
        table_name = arguments["table_name"]
        if not table_name.isidentifier():
            return CallToolResult(
                content=[TextContent(type="text", text="Invalid table name")],
                isError=True
            )

        async with pool.acquire() as conn:
            columns = await conn.fetch("""
                SELECT column_name, data_type, is_nullable, column_default
                FROM information_schema.columns
                WHERE table_name = $1
                ORDER BY ordinal_position
            """, table_name)

        schema = [dict(col) for col in columns]
        return CallToolResult(
            content=[TextContent(
                type="text",
                text=json.dumps(schema, indent=2)
            )]
        )

    raise ValueError(f"Unknown tool: {name}")

async def main():
    await init_db()
    async with stdio_server() as (read_stream, write_stream):
        await server.run(
            read_stream,
            write_stream,
            server.create_initialization_options()
        )

if __name__ == "__main__":
    asyncio.run(main())

Configuring MCP with Claude Desktop

Configuration File

Create or edit ~/Library/Application Support/Claude/claude_desktop_config.json (macOS) or %APPDATA%\Claude\claude_desktop_config.json (Windows):

{
  "mcpServers": {
    "demo-server": {
      "command": "python",
      "args": ["/path/to/mcp-server-demo/src/server.py"],
      "env": {
        "PYTHONPATH": "/path/to/mcp-server-demo"
      }
    },
    "postgres": {
      "command": "python",
      "args": ["/path/to/mcp-server-demo/src/postgres_server.py"],
      "env": {
        "DATABASE_URL": "postgresql://user:pass@localhost/db"
      }
    },
    "typescript-server": {
      "command": "node",
      "args": ["/path/to/mcp-ts-server/dist/server.js"]
    }
  }
}

Environment Variables

{
  "mcpServers": {
    "api-server": {
      "command": "python",
      "args": ["server.py"],
      "env": {
        "API_KEY": "your-api-key",
        "DEBUG": "true"
      }
    }
  }
}

Advanced Patterns

Streaming Responses

from mcp.types import TextContent

@server.call_tool()
async def call_tool(name: str, arguments: dict):
    if name == "analyze_large_dataset":
        # For large operations, provide progress updates
        results = []

        async for batch in process_data_batches(arguments["dataset"]):
            results.append(batch)

        return CallToolResult(
            content=[TextContent(
                type="text",
                text=json.dumps({
                    "status": "complete",
                    "processed": len(results),
                    "summary": generate_summary(results)
                })
            )]
        )

Error Handling

from mcp.types import CallToolResult, TextContent

@server.call_tool()
async def call_tool(name: str, arguments: dict) -> CallToolResult:
    try:
        result = await perform_operation(arguments)
        return CallToolResult(
            content=[TextContent(type="text", text=result)]
        )
    except ValidationError as e:
        return CallToolResult(
            content=[TextContent(type="text", text=f"Validation error: {e}")],
            isError=True
        )
    except DatabaseError as e:
        return CallToolResult(
            content=[TextContent(type="text", text=f"Database error: {e}")],
            isError=True
        )
    except Exception as e:
        return CallToolResult(
            content=[TextContent(type="text", text=f"Unexpected error: {e}")],
            isError=True
        )

Authentication and Security

import os
import hashlib
import hmac

class SecureMCPServer:
    def __init__(self):
        self.api_key = os.environ.get("MCP_API_KEY")
        self.allowed_operations = {"read", "list", "search"}

    def validate_request(self, operation: str, signature: str = None):
        if operation not in self.allowed_operations:
            raise PermissionError(f"Operation '{operation}' not allowed")

        if self.api_key and signature:
            expected = hmac.new(
                self.api_key.encode(),
                operation.encode(),
                hashlib.sha256
            ).hexdigest()
            if not hmac.compare_digest(signature, expected):
                raise PermissionError("Invalid signature")

    def sanitize_input(self, value: str) -> str:
        # Remove potentially dangerous characters
        dangerous = ["'", '"', ";", "--", "/*", "*/"]
        for char in dangerous:
            value = value.replace(char, "")
        return value.strip()

Testing MCP Servers

Unit Testing

# tests/test_server.py
import pytest
import json
from unittest.mock import AsyncMock, patch
from src.server import list_resources, list_tools, call_tool

@pytest.mark.asyncio
async def test_list_resources():
    resources = await list_resources()
    assert len(resources) == 2
    assert any(r.uri == "data://users" for r in resources)

@pytest.mark.asyncio
async def test_list_tools():
    tools = await list_tools()
    tool_names = [t.name for t in tools]
    assert "add_user" in tool_names
    assert "search_products" in tool_names

@pytest.mark.asyncio
async def test_add_user():
    result = await call_tool("add_user", {
        "name": "Test User",
        "email": "test@example.com"
    })
    assert "Created user" in result.content[0].text

@pytest.mark.asyncio
async def test_search_products():
    result = await call_tool("search_products", {"query": "Widget"})
    data = json.loads(result.content[0].text)
    assert len(data) > 0
    assert data[0]["name"] == "Widget"

Integration Testing

# tests/test_integration.py
import subprocess
import json

def test_server_starts():
    process = subprocess.Popen(
        ["python", "src/server.py"],
        stdin=subprocess.PIPE,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE
    )

    # Send initialization request
    init_request = {
        "jsonrpc": "2.0",
        "id": 1,
        "method": "initialize",
        "params": {
            "protocolVersion": "2024-11-05",
            "capabilities": {},
            "clientInfo": {"name": "test", "version": "1.0"}
        }
    }

    process.stdin.write(json.dumps(init_request).encode() + b"\n")
    process.stdin.flush()

    response = process.stdout.readline()
    result = json.loads(response)

    assert result["result"]["serverInfo"]["name"] == "demo-server"
    process.terminate()

Production Deployment

Docker Configuration

# Dockerfile
FROM python:3.11-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY src/ ./src/

ENV PYTHONUNBUFFERED=1

CMD ["python", "-m", "src.server"]

Docker Compose

# docker-compose.yml
version: '3.8'

services:
  mcp-server:
    build: .
    environment:
      - DATABASE_URL=postgresql://postgres:password@db:5432/mydb
      - MCP_API_KEY=${MCP_API_KEY}
    depends_on:
      - db

  db:
    image: postgres:15
    environment:
      - POSTGRES_PASSWORD=password
      - POSTGRES_DB=mydb
    volumes:
      - pgdata:/var/lib/postgresql/data

volumes:
  pgdata:

MCP Server Examples

GitHub Integration

import aiohttp
from mcp.server import Server
from mcp.types import Tool, TextContent, CallToolResult

server = Server("github-mcp")
GITHUB_TOKEN = os.environ.get("GITHUB_TOKEN")

@server.list_tools()
async def list_tools():
    return [
        Tool(
            name="list_repos",
            description="List GitHub repositories for a user",
            inputSchema={
                "type": "object",
                "properties": {
                    "username": {"type": "string"}
                },
                "required": ["username"]
            }
        ),
        Tool(
            name="get_issues",
            description="Get issues from a repository",
            inputSchema={
                "type": "object",
                "properties": {
                    "owner": {"type": "string"},
                    "repo": {"type": "string"},
                    "state": {"type": "string", "enum": ["open", "closed", "all"]}
                },
                "required": ["owner", "repo"]
            }
        )
    ]

@server.call_tool()
async def call_tool(name: str, arguments: dict) -> CallToolResult:
    headers = {
        "Authorization": f"token {GITHUB_TOKEN}",
        "Accept": "application/vnd.github.v3+json"
    }

    async with aiohttp.ClientSession(headers=headers) as session:
        if name == "list_repos":
            url = f"https://api.github.com/users/{arguments['username']}/repos"
            async with session.get(url) as resp:
                repos = await resp.json()
                summary = [{"name": r["name"], "stars": r["stargazers_count"]} for r in repos[:10]]

            return CallToolResult(
                content=[TextContent(type="text", text=json.dumps(summary, indent=2))]
            )

        elif name == "get_issues":
            owner, repo = arguments["owner"], arguments["repo"]
            state = arguments.get("state", "open")
            url = f"https://api.github.com/repos/{owner}/{repo}/issues?state={state}"

            async with session.get(url) as resp:
                issues = await resp.json()
                summary = [{"number": i["number"], "title": i["title"]} for i in issues[:20]]

            return CallToolResult(
                content=[TextContent(type="text", text=json.dumps(summary, indent=2))]
            )

Summary

Concept	Purpose
Resources	Expose data to LLMs
Tools	Enable LLM actions
Prompts	Reusable interaction templates
Server	Hosts MCP capabilities
Client	Connects LLM to servers

MCP enables seamless integration between AI models and external systems, making it easier to build powerful, context-aware AI applications.