MCP — Model Context Protocol

# mcp_server.py
from mcp.server import Server
from mcp.server.models import InitializationOptions
from mcp.server.stdio import stdio_server
from mcp import types
import asyncio
import sqlite3
import json

# Initialize the MCP server
app = Server("company-data-server")

# -------------------------------------------------------
# TOOLS — functions the AI can call
# -------------------------------------------------------
@app.list_tools()
async def list_tools() -> list[types.Tool]:
    return [
        types.Tool(
            name="query_employees",
            description=(
                "Query the employee database. Returns employee records matching "
                "the given filters. Use this to look up staff information."
            ),
            inputSchema={
                "type": "object",
                "properties": {
                    "department": {
                        "type": "string",
                        "description": "Filter by department (e.g. Engineering, Marketing)"
                    },
                    "min_years": {
                        "type": "integer",
                        "description": "Minimum years at company"
                    }
                },
            }
        ),
        types.Tool(
            name="send_notification",
            description="Send a Slack notification to a channel or user.",
            inputSchema={
                "type": "object",
                "properties": {
                    "channel": {"type": "string", "description": "Slack channel name"},
                    "message": {"type": "string", "description": "Message to send"},
                },
                "required": ["channel", "message"]
            }
        )
    ]

@app.call_tool()
async def call_tool(name: str, arguments: dict) -> list[types.TextContent]:
    if name == "query_employees":
        # Mock database query
        conn = sqlite3.connect(":memory:")
        # ... run query with arguments ...
        results = [{"name": "Alice", "dept": "Engineering", "years": 3}]
        return [types.TextContent(type="text", text=json.dumps(results, indent=2))]

    elif name == "send_notification":
        # Mock Slack call
        channel = arguments["channel"]
        message = arguments["message"]
        print(f"[Slack] #{channel}: {message}")
        return [types.TextContent(type="text", text=f"Notification sent to #{channel}")]

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

# -------------------------------------------------------
# RESOURCES — data the AI can read
# -------------------------------------------------------
@app.list_resources()
async def list_resources() -> list[types.Resource]:
    return [
        types.Resource(
            uri="company://docs/onboarding",
            name="Onboarding Guide",
            description="Employee onboarding documentation",
            mimeType="text/markdown",
        ),
        types.Resource(
            uri="company://policies/expense",
            name="Expense Policy",
            description="Company expense reimbursement policy",
            mimeType="text/markdown",
        )
    ]

@app.read_resource()
async def read_resource(uri: str) -> str:
    if uri == "company://docs/onboarding":
        return "# Onboarding Guide\n\nWelcome to the team!..."
    elif uri == "company://policies/expense":
        return "# Expense Policy\n\nReimbursable expenses include..."
    raise ValueError(f"Unknown resource: {uri}")

# -------------------------------------------------------
# Run the server over stdio transport
# -------------------------------------------------------
async def main():
    async with stdio_server() as (read_stream, write_stream):
        await app.run(
            read_stream,
            write_stream,
            InitializationOptions(
                server_name="company-data-server",
                server_version="1.0.0",
                capabilities=app.get_capabilities(
                    notification_options=None,
                    experimental_capabilities={}
                )
            )
        )

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

# mcp_client.py
import asyncio
from anthropic import Anthropic
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

client = Anthropic()

async def run_agent_with_mcp(user_message: str) -> str:
    """Run Claude as an agent connected to an MCP server."""

    # Connect to the MCP server via stdio
    server_params = StdioServerParameters(
        command="python",
        args=["mcp_server.py"],
    )

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

            # Discover available tools from the server
            tools_response = await session.list_tools()
            tools = [
                {
                    "name": t.name,
                    "description": t.description,
                    "input_schema": t.inputSchema,
                }
                for t in tools_response.tools
            ]
            print(f"Connected to MCP server. Available tools: {[t['name'] for t in tools]}")

            # Run the agentic loop
            messages = [{"role": "user", "content": user_message}]

            while True:
                response = client.messages.create(
                    model="claude-sonnet-4-6",
                    max_tokens=4096,
                    tools=tools,
                    messages=messages,
                )

                if response.stop_reason == "end_turn":
                    for block in response.content:
                        if hasattr(block, "text"):
                            return block.text

                # Execute tool calls via MCP
                messages.append({"role": "assistant", "content": response.content})
                tool_results = []

                for block in response.content:
                    if block.type == "tool_use":
                        # Call the tool on the MCP server
                        result = await session.call_tool(block.name, block.input)
                        tool_results.append({
                            "type": "tool_result",
                            "tool_use_id": block.id,
                            "content": result.content[0].text if result.content else "",
                        })

                if tool_results:
                    messages.append({"role": "user", "content": tool_results})

async def main():
    answer = await run_agent_with_mcp(
        "List all engineers with more than 2 years at the company, "
        "then send a Slack notification to #general about the team."
    )
    print(answer)

asyncio.run(main())