┌─────────────────┐     MCP Protocol      ┌─────────────────┐
│   AI Client     │◄────────────────────►│   MCP Server    │
│   (Claude,      │     (JSON-RPC)        │   (Your tools)  │
│    ChatGPT)     │                       │                 │
└─────────────────┘                       └─────────────────┘
                                                  │
                                                  ▼
                                          ┌─────────────────┐
                                          │ External Systems│
                                          │ (APIs, DBs,     │
                                          │  Files, etc.)   │
                                          └─────────────────┘

pip install mcp

# weather_server.py
from mcp.server import Server
from mcp.types import Tool, TextContent
import httpx

# Create the server
server = Server("weather-server")

# Define a tool
@server.tool()
async def get_weather(city: str) -> str:
    """Get current weather for a city.

    Args:
        city: Name of the city to get weather for
    """
    # In production, use a real weather API
    async with httpx.AsyncClient() as client:
        response = await client.get(
            f"https://wttr.in/{city}?format=j1"
        )
        data = response.json()

        current = data["current_condition"][0]
        return f"""Weather in {city}:
Temperature: {current['temp_C']}°C
Condition: {current['weatherDesc'][0]['value']}
Humidity: {current['humidity']}%
Wind: {current['windspeedKmph']} km/h"""

# Run the server
if __name__ == "__main__":
    import asyncio
    from mcp.server.stdio import stdio_server

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

    asyncio.run(main())

{
  "mcpServers": {
    "weather": {
      "command": "python",
      "args": ["/path/to/weather_server.py"]
    }
  }
}

# database_server.py
from mcp.server import Server
from mcp.types import Tool, Resource, TextContent
import sqlite3
from pathlib import Path

server = Server("database-server")
DB_PATH = Path("./data.db")

@server.list_resources()
async def list_resources():
    """List available database tables as resources."""
    conn = sqlite3.connect(DB_PATH)
    cursor = conn.execute(
        "SELECT name FROM sqlite_master WHERE type='table'"
    )
    tables = cursor.fetchall()
    conn.close()

    return [
        Resource(
            uri=f"db://tables/{table[0]}",
            name=f"Table: {table[0]}",
            description=f"Database table {table[0]}",
            mimeType="application/json"
        )
        for table in tables
    ]

@server.read_resource()
async def read_resource(uri: str):
    """Read table schema and sample data."""
    table_name = uri.split("/")[-1]

    conn = sqlite3.connect(DB_PATH)

    # Get schema
    cursor = conn.execute(f"PRAGMA table_info({table_name})")
    columns = cursor.fetchall()

    # Get sample data
    cursor = conn.execute(f"SELECT * FROM {table_name} LIMIT 5")
    sample = cursor.fetchall()

    conn.close()

    schema = [{"name": col[1], "type": col[2]} for col in columns]

    return TextContent(
        text=f"Schema: {schema}\nSample: {sample}"
    )

@server.tool()
async def query_database(sql: str) -> str:
    """Execute a SQL query and return results.

    Args:
        sql: SQL query to execute (SELECT only for safety)

    Returns:
        Query results as formatted text
    """
    # Security: Only allow SELECT queries
    if not sql.strip().upper().startswith("SELECT"):
        return "Error: Only SELECT queries are allowed"

    try:
        conn = sqlite3.connect(DB_PATH)
        conn.row_factory = sqlite3.Row
        cursor = conn.execute(sql)
        rows = cursor.fetchall()
        conn.close()

        if not rows:
            return "No results found"

        # Format as table
        headers = rows[0].keys()
        result = " | ".join(headers) + "\n"
        result += "-" * len(result) + "\n"

        for row in rows[:100]:  # Limit results
            result += " | ".join(str(row[h]) for h in headers) + "\n"

        return result

    except Exception as e:
        return f"Query error: {str(e)}"

@server.tool()
async def get_table_stats(table_name: str) -> str:
    """Get statistics about a database table.

    Args:
        table_name: Name of the table to analyze
    """
    conn = sqlite3.connect(DB_PATH)

    # Row count
    cursor = conn.execute(f"SELECT COUNT(*) FROM {table_name}")
    count = cursor.fetchone()[0]

    # Column info
    cursor = conn.execute(f"PRAGMA table_info({table_name})")
    columns = cursor.fetchall()

    conn.close()

    stats = f"Table: {table_name}\n"
    stats += f"Rows: {count}\n"
    stats += f"Columns: {len(columns)}\n\n"
    stats += "Column Details:\n"

    for col in columns:
        stats += f"  - {col[1]} ({col[2]})\n"

    return stats

@server.list_prompts()
async def list_prompts():
    return [
        Prompt(
            name="analyze-data",
            description="Analyze a dataset with specific focus",
            arguments=[
                PromptArgument(
                    name="focus",
                    description="What aspect to focus on",
                    required=True
                )
            ]
        )
    ]

@server.get_prompt()
async def get_prompt(name: str, arguments: dict):
    if name == "analyze-data":
        focus = arguments.get("focus", "general patterns")
        return PromptResult(
            messages=[
                PromptMessage(
                    role="user",
                    content=TextContent(
                        text=f"""Please analyze the data with a focus on {focus}.

Consider:
1. Key patterns and trends
2. Anomalies or outliers
3. Actionable insights
4. Potential issues or data quality concerns

Provide a structured analysis with clear sections."""
                    )
                )
            ]
        )

from pydantic import BaseModel, validator

class QueryInput(BaseModel):
    sql: str

    @validator("sql")
    def validate_sql(cls, v):
        # Prevent dangerous operations
        dangerous = ["DROP", "DELETE", "UPDATE", "INSERT", "ALTER", "CREATE"]
        if any(word in v.upper() for word in dangerous):
            raise ValueError("Only SELECT queries allowed")
        return v

@server.tool()
async def safe_query(query: QueryInput) -> str:
    # Input is validated by Pydantic
    return execute_query(query.sql)

from datetime import datetime, timedelta
from collections import defaultdict

class RateLimiter:
    def __init__(self, max_requests: int, window_seconds: int):
        self.max_requests = max_requests
        self.window = timedelta(seconds=window_seconds)
        self.requests = defaultdict(list)

    def is_allowed(self, key: str) -> bool:
        now = datetime.now()
        cutoff = now - self.window

        # Clean old requests
        self.requests[key] = [
            t for t in self.requests[key] if t > cutoff
        ]

        if len(self.requests[key]) >= self.max_requests:
            return False

        self.requests[key].append(now)
        return True

limiter = RateLimiter(max_requests=100, window_seconds=60)

@server.tool()
async def rate_limited_tool(input: str) -> str:
    if not limiter.is_allowed("default"):
        return "Rate limit exceeded. Please wait."
    # Process request

from enum import Enum

class Permission(Enum):
    READ = "read"
    WRITE = "write"
    ADMIN = "admin"

class PermissionManager:
    def __init__(self):
        self.permissions = {
            "read_data": Permission.READ,
            "query_database": Permission.READ,
            "modify_data": Permission.WRITE,
            "delete_data": Permission.ADMIN,
        }

    def check(self, tool_name: str, user_permission: Permission) -> bool:
        required = self.permissions.get(tool_name, Permission.ADMIN)
        return user_permission.value >= required.value

# test_weather_server.py
import pytest
from mcp.client import ClientSession
from mcp.client.stdio import stdio_client

@pytest.fixture
async def client():
    async with stdio_client(
        command="python",
        args=["weather_server.py"]
    ) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()
            yield session

@pytest.mark.asyncio
async def test_list_tools(client):
    tools = await client.list_tools()
    assert any(t.name == "get_weather" for t in tools)

@pytest.mark.asyncio
async def test_get_weather(client):
    result = await client.call_tool("get_weather", {"city": "London"})
    assert "Temperature" in result.content[0].text

{
  "mcpServers": {
    "filesystem": {
      "command": "npx",
      "args": ["@modelcontextprotocol/server-filesystem", "/path/to/allowed/directory"]
    }
  }
}

from mcp.client import ClientSession
from mcp.client.stdio import stdio_client

async def use_mcp_tools():
    # Connect to MCP server
    async with stdio_client(
        command="python",
        args=["my_server.py"]
    ) as (read, write):
        async with ClientSession(read, write) as session:
            # Initialize connection
            await session.initialize()

            # Discover available tools
            tools = await session.list_tools()
            print(f"Available tools: {[t.name for t in tools]}")

            # Call a tool
            result = await session.call_tool(
                "get_weather",
                {"city": "Tokyo"}
            )
            print(result.content[0].text)

            # Read a resource
            resources = await session.list_resources()
            if resources:
                content = await session.read_resource(resources[0].uri)
                print(content)