Python MCP Server Development: Building AI Agent-Callable Tool Services from Scratch in 2026

When developing AI Agents, have you noticed that every AI platform has its own tool-calling protocol? Claude uses one, ChatGPT uses another, and Gemini is different again? You have to write adapter layers for each platform? In 2026, Model Context Protocol (MCP) has become the open standard for AI tool calling — write an MCP Server once, and all major AI platforms can call it.

---

Background

MCP Protocol Architecture

MCP (Model Context Protocol) is an open protocol released by Anthropic in late 2024 that defines the standard communication method between AI applications (Host) and tool services (Server).

Concept	Description	Analogy
Host	AI application (Claude Desktop, ChatGPT, etc.)	Browser
Client	MCP client inside the Host	HTTP Client
Server	Service providing tools/resources/prompts	HTTP Server
Transport	Communication method (stdio/SSE/Streamable HTTP)	TCP/HTTP

MCP Core Capabilities

Capability	Description	Example
Tools	Functions callable by AI	Query database, send email
Resources	Data readable by AI	File contents, database records
Prompts	Predefined prompt templates	Code review template
Sampling	AI requests Host to generate	Intermediate steps requiring AI judgment

Transport Comparison

Transport	Use Case	Pros	Cons
stdio	Local desktop apps	Simple, secure	Local process only
SSE	Remote web services	Cross-network	Long connection maintenance
Streamable HTTP	Remote web services (recommended)	Stateless, CDN-friendly	Requires MCP 2025-03 spec

---

Problem Analysis

Why Do We Need MCP?

1. Protocol fragmentation: Each AI platform has a different tool-calling format, high adaptation cost
2. Tool reuse difficulty: Tools written for Claude cannot be used in ChatGPT
3. Security risks: Directly exposing API endpoints to AI lacks permission control
4. Chaotic state management: No standard for context passing between AI and tools

MCP solves these problems through a unified protocol: one MCP Server can be called by any AI Host that supports MCP.

---

Step-by-Step Guide

Step 1: Install MCP Python SDK

pip install mcp[cli]

Step 2: Create Your First MCP Server

# server.py
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import Tool, TextContent

app = Server("toolsku-tools")

@app.list_tools()
async def list_tools() -> list[Tool]:
    return [
        Tool(
            name="calculate",
            description="Execute mathematical expression calculation",
            inputSchema={
                "type": "object",
                "properties": {
                    "expression": {
                        "type": "string",
                        "description": "Mathematical expression, e.g. '2 + 3 * 4'",
                    }
                },
                "required": ["expression"],
            },
        ),
        Tool(
            name="format_json",
            description="Format JSON string",
            inputSchema={
                "type": "object",
                "properties": {
                    "json_str": {
                        "type": "string",
                        "description": "JSON string to format",
                    },
                    "indent": {
                        "type": "integer",
                        "description": "Number of indent spaces, default 2",
                        "default": 2,
                    },
                },
                "required": ["json_str"],
            },
        ),
    ]

@app.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
    if name == "calculate":
        try:
            result = eval(arguments["expression"], {"__builtins__": {}}, {})
            return [TextContent(type="text", text=f"Result: {result}")]
        except Exception as e:
            return [TextContent(type="text", text=f"Calculation error: {str(e)}")]

    elif name == "format_json":
        import json
        try:
            parsed = json.loads(arguments["json_str"])
            indent = arguments.get("indent", 2)
            formatted = json.dumps(parsed, indent=indent, ensure_ascii=False)
            return [TextContent(type="text", text=formatted)]
        except json.JSONDecodeError as e:
            return [TextContent(type="text", text=f"JSON parse error: {str(e)}")]

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

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

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

Step 3: Configure Claude Desktop Connection

// ~/Library/Application Support/Claude/claude_desktop_config.json (macOS)
// %APPDATA%\Claude\claude_desktop_config.json (Windows)
{
  "mcpServers": {
    "toolsku-tools": {
      "command": "python",
      "args": ["server.py"],
      "cwd": "E:\\project\\toolsku\\mcp-server"
    }
  }
}

Step 4: SSE Transport Mode (Remote Deployment)

# sse_server.py
from mcp.server import Server
from mcp.server.sse import SseServerTransport
from mcp.types import Tool, TextContent
from starlette.applications import Starlette
from starlette.routing import Mount, Route

app = Server("toolsku-remote-tools")
sse = SseServerTransport("/messages/")

@app.list_tools()
async def list_tools() -> list[Tool]:
    return [
        Tool(
            name="search_docs",
            description="Search technical documentation",
            inputSchema={
                "type": "object",
                "properties": {
                    "query": {"type": "string", "description": "Search keyword"},
                    "category": {"type": "string", "description": "Document category"},
                    "limit": {"type": "integer", "description": "Number of results", "default": 5},
                },
                "required": ["query"],
            },
        ),
    ]

@app.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
    if name == "search_docs":
        results = await search_documents(
            query=arguments["query"],
            category=arguments.get("category"),
            limit=arguments.get("limit", 5),
        )
        import json
        return [TextContent(type="text", text=json.dumps(results, ensure_ascii=False, indent=2))]
    raise ValueError(f"Unknown tool: {name}")

async def search_documents(query: str, category: str | None, limit: int) -> list[dict]:
    return [
        {"title": f"Doc: {query}", "url": f"https://docs.example.com/{query}", "score": 0.95},
    ][:limit]

async def handle_sse(request):
    async with sse.connect_sse(request.scope, request.receive, request._send) as streams:
        await app.run(streams[0], streams[1], app.create_initialization_options())

starlette_app = Starlette(
    routes=[
        Route("/sse", endpoint=handle_sse),
        Mount("/messages/", app=sse.handle_post_message),
    ],
)

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(starlette_app, host="0.0.0.0", port=8080)

Step 5: Integrate with ChatGPT

# chatgpt_adapter.py
import httpx
from openai import OpenAI

MCP_SERVER_URL = "http://localhost:8080"

async def discover_mcp_tools() -> list[dict]:
    async with httpx.AsyncClient() as client:
        response = await client.post(
            f"{MCP_SERVER_URL}/messages/",
            json={
                "jsonrpc": "2.0",
                "method": "tools/list",
                "id": 1,
            },
        )
        result = response.json()
        return result.get("result", {}).get("tools", [])

def mcp_tools_to_openai_tools(mcp_tools: list[dict]) -> list[dict]:
    openai_tools = []
    for tool in mcp_tools:
        openai_tools.append({
            "type": "function",
            "function": {
                "name": tool["name"],
                "description": tool["description"],
                "parameters": tool["inputSchema"],
            },
        })
    return openai_tools

async def call_mcp_tool(name: str, arguments: dict) -> str:
    async with httpx.AsyncClient() as client:
        response = await client.post(
            f"{MCP_SERVER_URL}/messages/",
            json={
                "jsonrpc": "2.0",
                "method": "tools/call",
                "params": {"name": name, "arguments": arguments},
                "id": 2,
            },
        )
        result = response.json()
        contents = result.get("result", {}).get("content", [])
        return "\n".join(c.get("text", "") for c in contents)

async def chat_with_tools(user_message: str) -> str:
    mcp_tools = await discover_mcp_tools()
    openai_tools = mcp_tools_to_openai_tools(mcp_tools)

    client = OpenAI()
    messages = [{"role": "user", "content": user_message}]

    response = client.chat.completions.create(
        model="gpt-4o",
        messages=messages,
        tools=openai_tools,
    )

    message = response.choices[0].message

    if message.tool_calls:
        for tool_call in message.tool_calls:
            tool_result = await call_mcp_tool(
                tool_call.function.name,
                json.loads(tool_call.function.arguments),
            )
            messages.append(message)
            messages.append({
                "role": "tool",
                "tool_call_id": tool_call.id,
                "content": tool_result,
            })

        response = client.chat.completions.create(
            model="gpt-4o",
            messages=messages,
            tools=openai_tools,
        )
        return response.choices[0].message.content

    return message.content

---

Complete Code: Production-Grade MCP Server

# production_server.py
import os
import json
import logging
from datetime import datetime
from typing import Any

from mcp.server import Server
from mcp.server.sse import SseServerTransport
from mcp.types import Tool, TextContent, ImageContent
from starlette.applications import Starlette
from starlette.routing import Mount, Route
from starlette.middleware import Middleware
from starlette.middleware.cors import CORSMiddleware

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("toolsku-mcp")

app = Server("toolsku-production")

API_KEY = os.environ.get("MCP_API_KEY", "dev-key")

def validate_api_key(headers: dict) -> bool:
    auth = headers.get("authorization", "")
    return auth == f"Bearer {API_KEY}"

@app.list_tools()
async def list_tools() -> list[Tool]:
    return [
        Tool(
            name="base64_encode",
            description="Base64 encode a string",
            inputSchema={
                "type": "object",
                "properties": {
                    "text": {"type": "string", "description": "Text to encode"},
                },
                "required": ["text"],
            },
        ),
        Tool(
            name="base64_decode",
            description="Base64 decode a string",
            inputSchema={
                "type": "object",
                "properties": {
                    "encoded": {"type": "string", "description": "Base64 encoded string"},
                },
                "required": ["encoded"],
            },
        ),
        Tool(
            name="json_format",
            description="Format JSON string",
            inputSchema={
                "type": "object",
                "properties": {
                    "json_str": {"type": "string", "description": "JSON string"},
                    "indent": {"type": "integer", "description": "Indentation", "default": 2},
                    "sort_keys": {"type": "boolean", "description": "Whether to sort keys", "default": False},
                },
                "required": ["json_str"],
            },
        ),
        Tool(
            name="hash_compute",
            description="Compute hash of text",
            inputSchema={
                "type": "object",
                "properties": {
                    "text": {"type": "string", "description": "Text to hash"},
                    "algorithm": {
                        "type": "string",
                        "enum": ["md5", "sha1", "sha256", "sha512"],
                        "description": "Hash algorithm",
                        "default": "sha256",
                    },
                },
                "required": ["text"],
            },
        ),
        Tool(
            name="timestamp_convert",
            description="Convert between timestamp and date format",
            inputSchema={
                "type": "object",
                "properties": {
                    "value": {"type": "string", "description": "Timestamp or date string"},
                    "format": {"type": "string", "description": "Target format", "default": "iso"},
                },
                "required": ["value"],
            },
        ),
    ]

@app.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
    logger.info(f"Tool called: {name} with args: {arguments}")

    try:
        if name == "base64_encode":
            import base64
            result = base64.b64encode(arguments["text"].encode()).decode()
            return [TextContent(type="text", text=result)]

        elif name == "base64_decode":
            import base64
            result = base64.b64decode(arguments["encoded"]).decode()
            return [TextContent(type="text", text=result)]

        elif name == "json_format":
            parsed = json.loads(arguments["json_str"])
            indent = arguments.get("indent", 2)
            sort_keys = arguments.get("sort_keys", False)
            formatted = json.dumps(parsed, indent=indent, sort_keys=sort_keys, ensure_ascii=False)
            return [TextContent(type="text", text=formatted)]

        elif name == "hash_compute":
            import hashlib
            algorithm = arguments.get("algorithm", "sha256")
            h = hashlib.new(algorithm)
            h.update(arguments["text"].encode())
            return [TextContent(type="text", text=h.hexdigest())]

        elif name == "timestamp_convert":
            value = arguments["value"]
            fmt = arguments.get("format", "iso")
            try:
                ts = float(value)
                dt = datetime.fromtimestamp(ts)
            except ValueError:
                dt = datetime.fromisoformat(value)
                ts = dt.timestamp()

            if fmt == "iso":
                result = dt.isoformat()
            elif fmt == "timestamp":
                result = str(ts)
            else:
                result = dt.strftime(fmt)

            return [TextContent(type="text", text=result)]

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

    except Exception as e:
        logger.error(f"Tool error: {name} - {str(e)}")
        return [TextContent(type="text", text=f"Error: {str(e)}")]

sse = SseServerTransport("/messages/")

async def handle_sse(request):
    async with sse.connect_sse(request.scope, request.receive, request._send) as streams:
        await app.run(streams[0], streams[1], app.create_initialization_options())

middleware = [
    Middleware(CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"]),
]

starlette_app = Starlette(
    routes=[
        Route("/sse", endpoint=handle_sse),
        Mount("/messages/", app=sse.handle_post_message),
    ],
    middleware=middleware,
)

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(starlette_app, host="0.0.0.0", port=8080)

---

Pitfall Guide

Pitfall 1: Claude Desktop Cannot Discover MCP Server

Symptom: Configured claude_desktop_config.json but Claude Desktop doesn't see the tools.

Solution: Check if the JSON format is correct (watch for commas and quotes), confirm the command path is correct (use absolute paths), check Claude Desktop logs (Help > Toggle Developer Tools).

Pitfall 2: SSE Connection Frequently Disconnects

Symptom: SSE connection to remote MCP Server drops every 30 seconds.

Solution: Ensure the server implements a heartbeat mechanism (send keep-alive every 15 seconds), configure the reverse proxy timeout to be greater than 60 seconds. Nginx requires proxy_read_timeout 300s.

Pitfall 3: Loose Tool Parameter Validation Causes Runtime Errors

Symptom: AI passes arguments of the wrong type (e.g., a string where an integer is expected).

Solution: Strictly define types and constraints in inputSchema (type, minimum, enum, etc.), add parameter validation and type conversion in the call_tool function.

Pitfall 4: Inconsistent Python Environment in stdio Mode

Symptom: Claude Desktop cannot find installed packages when starting the MCP Server.

Solution: Use the absolute path of the Python interpreter in claude_desktop_config.json, or use uv run to specify a virtual environment:

{
  "mcpServers": {
    "toolsku-tools": {
      "command": "uv",
      "args": ["run", "--directory", "E:\\project\\toolsku\\mcp-server", "server.py"]
    }
  }
}

Pitfall 5: Concurrent Requests Cause State Confusion

Symptom: Multiple AIs call the same MCP Server simultaneously, shared state gets overwritten.

Solution: Design the MCP Server to be stateless, passing all state through parameters. If state is necessary, use request-level context isolation and avoid global variables.

---

Error Troubleshooting

#	Error Message	Cause	Solution
1	MCP server not found in config	Config file path or format error	Check JSON format, use absolute paths
2	Connection refused on stdio	Python process failed to start	Check command path, check stderr logs
3	SSE connection timeout	Server not responding or network issue	Check server status, increase timeout
4	Tool not found: xxx	Tool name typo	Ensure list_tools and call_tool names match
5	JSON-RPC error: -32600	Request format doesn't match JSON-RPC spec	Check request body format
6	Permission denied	Tool execution requires permissions	Check file/network permissions, use sudo carefully
7	ModuleNotFoundError	Python dependency not installed	Specify correct Python environment in MCP config
8	413 Payload Too Large	Request body too large	Limit tool parameter size, use pagination
9	CORS error	SSE cross-origin request blocked	Configure CORS middleware
10	Server initialization failed	MCP handshake failed	Check initialize response format and capabilities

---

Advanced Optimization

1. Resource Exposure

@app.list_resources()
async def list_resources() -> list[Resource]:
    return [
        Resource(
            uri="toolsku:///docs/api-reference",
            name="API Reference Documentation",
            mimeType="text/markdown",
        ),
    ]

@app.read_resource()
async def read_resource(uri: str) -> str:
    if uri == "toolsku:///docs/api-reference":
        return "# API Reference\n\n## Tool List\n..."
    raise ValueError(f"Unknown resource: {uri}")

2. Prompt Templates

@app.list_prompts()
async def list_prompts() -> list[Prompt]:
    return [
        Prompt(
            name="code-review",
            description="Code review prompt template",
            arguments=[
                PromptArgument(name="language", description="Programming language", required=True),
                PromptArgument(name="code", description="Code to review", required=True),
            ],
        ),
    ]

@app.get_prompt()
async def get_prompt(name: str, arguments: dict) -> list[PromptMessage]:
    if name == "code-review":
        return [
            PromptMessage(
                role="user",
                content=TextContent(
                    type="text",
                    text=f"Please review the following {arguments['language']} code, focusing on security, performance, and maintainability:\n\n```{arguments['language']}\n{arguments['code']}\n```",
                ),
            ),
        ]
    raise ValueError(f"Unknown prompt: {name}")

3. Streamable HTTP Transport (MCP 2025-03)

from mcp.server.streamable_http import StreamableHTTPTransport

http_transport = StreamableHTTPTransport("/mcp")

async def handle_http(request):
    await http_transport.handle_request(request, app)

4. Authentication and Rate Limiting

from starlette.middleware.base import BaseHTTPMiddleware

class AuthMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request, call_next):
        if request.url.path in ["/sse", "/messages/"]:
            auth = request.headers.get("authorization", "")
            if auth != f"Bearer {API_KEY}":
                from starlette.responses import JSONResponse
                return JSONResponse({"error": "unauthorized"}, status_code=401)
        return await call_next(request)

---

Comparison Analysis

Dimension	MCP	OpenAI Function Calling	LangChain Tools	LlamaIndex Tools
Protocol Standard	Open standard	Platform proprietary	Framework proprietary	Framework proprietary
Cross-platform	All MCP-supporting Hosts	OpenAI only	LangChain only	LlamaIndex only
Transport	stdio/SSE/HTTP	HTTP API	Python function	Python function
Tool Discovery	Dynamic list_tools	Static definition	Static definition	Static definition
Resource Exposure	Native support	Not supported	Not supported	Not supported
Security Model	Permission control	API Key	No standard	No standard
Deployment	Local/Remote	Remote	In-process	In-process
Ecosystem	Rapidly growing	Mature	Mature	Mature

---

Summary and Outlook

Summary: MCP has become the open standard for AI tool calling in 2026. The Python MCP SDK makes developing an MCP Server extremely simple — define tools, register handlers, choose a transport method. The core advantage is "develop once, call from multiple platforms": the same MCP Server can be used by Claude Desktop, ChatGPT, Cursor, and any other AI application that supports MCP. We recommend starting with simple tool services (e.g., JSON formatting, Base64 encoding/decoding), gradually adding resource and prompt template capabilities, and ultimately building a complete AI tool ecosystem.

---

Recommended Online Tools

• JSON Formatter (MCP tool development debugging): /en/json/format
• Base64 Encoder/Decoder (MCP tool functionality): /en/encode/base64
• curl to Code (MCP API testing): /en/dev/curl-to-code