┌─────────────────┐        MCP Protocol         ┌─────────────────┐
│                 │◄──────────────────────────►│                 │
│  MCP Client     │     JSON-RPC Messages       │   MCP Server    │
│  (AI App/LLM)   │                             │  (Tools/Data)   │
│                 │                             │                 │
└─────────────────┘                             └─────────────────┘
        │                                                │
        │                                                │
    Uses tools,                                    Exposes:
    reads resources,                               - Tools
    gets prompts                                   - Resources
                                                   - Prompts

pip install mcp

from mcp.server.mcpserver import MCPServer

# Create server
mcp = MCPServer("My First Server")

# Add a tool
@mcp.tool()
def add(a: int, b: int) -> int:
    """Add two numbers"""
    return a + b

# Add a resource
@mcp.resource("greeting://{name}")
def get_greeting(name: str) -> str:
    """Get a personalized greeting"""
    return f"Hello, {name}!"

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

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

import asyncio
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

async def main():
    server_params = StdioServerParameters(
        command="python",
        args=["my_server.py"]
    )

    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()

            # Call a tool
            result = await session.call_tool("add", {"a": 5, "b": 3})
            print(result.content[0].text)  # "8"

            # Read a resource
            content = await session.read_resource("greeting://Alice")
            print(content.contents[0].text)  # "Hello, Alice!"

asyncio.run(main())

@mcp.tool()
def search_database(query: str, limit: int = 10) -> list[dict]:
    """Search the database.

    Args:
        query: Search query string
        limit: Maximum results to return
    """
    # Implementation
    results = db.search(query, limit=limit)
    return results

@mcp.resource("file://{path}")
def read_file(path: str) -> str:
    """Read file contents."""
    with open(path) as f:
        return f.read()

# Dynamic resources with URI templates
@mcp.resource("user://{user_id}/profile")
def get_user_profile(user_id: str) -> dict:
    """Get user profile data."""
    return db.get_user(user_id)

@mcp.prompt()
def analyze_logs(log_file: str, error_type: str = "all") -> str:
    """Generate log analysis prompt.

    Args:
        log_file: Path to log file
        error_type: Type of errors to focus on
    """
    return f"""Analyze the logs in {log_file}.
Focus on: {error_type}
Provide:
1. Error summary
2. Root causes
3. Recommendations"""

mcp.run(transport="stdio")

server_params = StdioServerParameters(
    command="python",
    args=["server.py"],
    env={"DEBUG": "true"}
)

async with stdio_client(server_params) as (read, write):
    async with ClientSession(read, write) as session:
        # Use session...

mcp.run(
    transport="streamable-http",
    stateless_http=True,
    json_response=True
)

from mcp.server.mcpserver import MCPServer

mcp = MCPServer("Demo Server", version="1.0.0")

@mcp.tool()
def my_tool(param: str) -> str:
    """Tool description"""
    return f"Result: {param}"

@mcp.resource("resource://uri")
def my_resource() -> str:
    """Resource description"""
    return "Resource content"

@mcp.prompt()
def my_prompt(arg: str) -> str:
    """Prompt description"""
    return f"Prompt: {arg}"

mcp.run(transport="stdio")

import asyncio
from mcp import ClientSession, StdioServerParameters, types
from mcp.client.stdio import stdio_client

async def use_mcp_server():
    server_params = StdioServerParameters(
        command="python",
        args=["my_server.py"]
    )

    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            # Initialize
            await session.initialize()

            # List available tools
            tools = await session.list_tools()

            # Call a tool
            result = await session.call_tool("add", {"a": 5, "b": 3})

            # Read resource
            content = await session.read_resource("greeting://World")

from mcp.server.mcpserver import Context

@mcp.tool()
async def long_task(ctx: Context, steps: int = 10) -> str:
    """Long running task with progress."""
    for i in range(steps):
        await ctx.report_progress(
            progress=(i + 1) / steps,
            total=1.0,
            message=f"Step {i + 1}/{steps}"
        )
    return "Complete"

@mcp.tool()
async def process_data(data: str, ctx: Context) -> str:
    """Process data with logging."""
    await ctx.debug(f"Processing: {data}")
    await ctx.info("Starting processing")
    await ctx.warning("This is experimental")
    return f"Processed: {data}"

from mcp.shared.exceptions import MCPError

try:
    result = await session.call_tool("my_tool", args)
except MCPError as e:
    print(f"MCP error: {e.message}")