MCP — Model Context Protocol

"""Complete MCP Server — Weather tool + Notes resource.
Install: pip install mcp
Run: python weather_server.py
"""
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import (
    Tool, TextContent, Resource, ResourceTemplate,
    GetPromptResult, PromptMessage
)
import json
import httpx
from datetime import datetime

# ─── Create the MCP Server ───────────────────────────────────

app = Server("weather-notes-server")

# ─── In-memory notes storage ─────────────────────────────────

notes_db: dict[str, dict] = {}

# ─── Tool: Get Weather ───────────────────────────────────────

@app.list_tools()
async def list_tools() -> list[Tool]:
    """Advertise available tools to MCP clients."""
    return [
        Tool(
            name="get_weather",
            description="Get current weather for a city. Returns temperature, conditions, and humidity.",
            inputSchema={
                "type": "object",
                "properties": {
                    "city": {
                        "type": "string",
                        "description": "City name (e.g., 'London', 'New York')"
                    },
                    "units": {
                        "type": "string",
                        "enum": ["celsius", "fahrenheit"],
                        "description": "Temperature units",
                        "default": "celsius"
                    }
                },
                "required": ["city"]
            }
        ),
        Tool(
            name="add_note",
            description="Save a note with a title and content. Notes persist for the session.",
            inputSchema={
                "type": "object",
                "properties": {
                    "title": {
                        "type": "string",
                        "description": "Note title (used as identifier)"
                    },
                    "content": {
                        "type": "string",
                        "description": "Note content (markdown supported)"
                    },
                    "tags": {
                        "type": "array",
                        "items": {"type": "string"},
                        "description": "Optional tags for categorization"
                    }
                },
                "required": ["title", "content"]
            }
        ),
        Tool(
            name="search_notes",
            description="Search notes by keyword in title or content.",
            inputSchema={
                "type": "object",
                "properties": {
                    "query": {
                        "type": "string",
                        "description": "Search keyword"
                    }
                },
                "required": ["query"]
            }
        ),
    ]

@app.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
    """Handle tool execution requests."""
    
    if name == "get_weather":
        city = arguments["city"]
        units = arguments.get("units", "celsius")
        
        # In production, call a real weather API
        # Demo: using Open-Meteo (free, no API key needed)
        try:
            async with httpx.AsyncClient() as client:
                # Geocode the city
                geo_resp = await client.get(
                    "https://geocoding-api.open-meteo.com/v1/search",
                    params={"name": city, "count": 1}
                )
                geo_data = geo_resp.json()
                
                if not geo_data.get("results"):
                    return [TextContent(type="text", text=f"City '{city}' not found.")]
                
                lat = geo_data["results"][0]["latitude"]
                lon = geo_data["results"][0]["longitude"]
                
                # Get weather
                unit_param = "celsius" if units == "celsius" else "fahrenheit"
                weather_resp = await client.get(
                    "https://api.open-meteo.com/v1/forecast",
                    params={
                        "latitude": lat, "longitude": lon,
                        "current": "temperature_2m,relative_humidity_2m,weather_code",
                        "temperature_unit": unit_param,
                    }
                )
                weather = weather_resp.json()["current"]
                
                result = (
                    f"Weather in {city}:\n"
                    f"  Temperature: {weather['temperature_2m']}°{'C' if units == 'celsius' else 'F'}\n"
                    f"  Humidity: {weather['relative_humidity_2m']}%\n"
                    f"  Conditions: WMO code {weather['weather_code']}"
                )
                return [TextContent(type="text", text=result)]
        
        except Exception as e:
            return [TextContent(type="text", text=f"Weather lookup failed: {e}")]
    
    elif name == "add_note":
        title = arguments["title"]
        content = arguments["content"]
        tags = arguments.get("tags", [])
        
        notes_db[title] = {
            "content": content,
            "tags": tags,
            "created": datetime.now().isoformat(),
        }
        return [TextContent(type="text", text=f"Note '{title}' saved successfully.")]
    
    elif name == "search_notes":
        query = arguments["query"].lower()
        matches = []
        for title, note in notes_db.items():
            if query in title.lower() or query in note["content"].lower():
                matches.append(f"- **{title}**: {note['content'][:100]}...")
        
        if matches:
            return [TextContent(type="text", text=f"Found {len(matches)} notes:\n" + "\n".join(matches))]
        return [TextContent(type="text", text="No matching notes found.")]
    
    return [TextContent(type="text", text=f"Unknown tool: {name}")]

# ─── Resources: Expose notes as readable resources ───────────

@app.list_resources()
async def list_resources() -> list[Resource]:
    """List all notes as resources."""
    resources = []
    for title, note in notes_db.items():
        resources.append(Resource(
            uri=f"notes://{title.replace(' ', '-').lower()}",
            name=title,
            description=f"Note: {title} (tags: {', '.join(note['tags'])})",
            mimeType="text/plain",
        ))
    return resources

@app.read_resource()
async def read_resource(uri: str) -> str:
    """Read a specific note by URI."""
    # Extract title from URI
    slug = uri.replace("notes://", "")
    for title, note in notes_db.items():
        if title.replace(" ", "-").lower() == slug:
            return note["content"]
    raise ValueError(f"Resource not found: {uri}")

# ─── Run the Server ──────────────────────────────────────────

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())

# Claude Desktop config file location:
# macOS: ~/Library/Application Support/Claude/claude_desktop_config.json
# Windows: %APPDATA%\Claude\claude_desktop_config.json

# Configuration:
{
    "mcpServers": {
        "weather-notes": {
            "command": "python",
            "args": ["/path/to/weather_server.py"],
            "env": {
                "WEATHER_API_KEY": "optional-api-key"
            }
        },
        "filesystem": {
            "command": "npx",
            "args": ["-y", "@modelcontextprotocol/server-filesystem", "/Users/you/Documents"]
        },
        "github": {
            "command": "npx",
            "args": ["-y", "@modelcontextprotocol/server-github"],
            "env": {
                "GITHUB_PERSONAL_ACCESS_TOKEN": "ghp_..."
            }
        }
    }
}

"""Testing MCP servers programmatically."""
from mcp.client import ClientSession
from mcp.client.stdio import stdio_client, StdioServerParameters
import asyncio

async def test_server():
    """Connect to and test an MCP server."""
    
    # Start the server as a subprocess
    server_params = StdioServerParameters(
        command="python",
        args=["weather_server.py"],
    )
    
    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            # Initialize the connection
            await session.initialize()
            
            # List available tools
            tools = await session.list_tools()
            print(f"Available tools: {[t.name for t in tools.tools]}")
            
            # Call a tool
            result = await session.call_tool("get_weather", {"city": "London"})
            print(f"Weather result: {result.content[0].text}")
            
            # Add a note
            result = await session.call_tool("add_note", {
                "title": "Meeting Notes",
                "content": "Discussed MCP server architecture",
                "tags": ["work", "mcp"]
            })
            print(f"Note result: {result.content[0].text}")
            
            # List resources
            resources = await session.list_resources()
            print(f"Resources: {[r.name for r in resources.resources]}")
            
            # Search notes
            result = await session.call_tool("search_notes", {"query": "MCP"})
            print(f"Search result: {result.content[0].text}")

asyncio.run(test_server())

"""MCP Inspector — use the official debugging tool.
Install: npm install -g @modelcontextprotocol/inspector
Run:     mcp-inspector python weather_server.py

This opens a web UI where you can:
- See all tools, resources, and prompts
- Call tools interactively
- View request/response JSON
- Test error handling
"""

# You can also test with the MCP CLI:
# pip install mcp[cli]
# mcp dev weather_server.py

"""MCP Server exposing a SQLite database as tools + resources."""
from mcp.server import Server
from mcp.types import Tool, TextContent, Resource
import sqlite3
import json

app = Server("sqlite-server")
DB_PATH = "app_data.db"

def get_db():
    return sqlite3.connect(DB_PATH)

@app.list_tools()
async def list_tools() -> list[Tool]:
    return [
        Tool(
            name="query_database",
            description="Run a READ-ONLY SQL query against the database. "
                       "Only SELECT statements are allowed.",
            inputSchema={
                "type": "object",
                "properties": {
                    "sql": {
                        "type": "string",
                        "description": "SQL SELECT query to execute"
                    }
                },
                "required": ["sql"]
            }
        ),
        Tool(
            name="list_tables",
            description="List all tables in the database with their schemas.",
            inputSchema={"type": "object", "properties": {}}
        ),
    ]

@app.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
    if name == "query_database":
        sql = arguments["sql"].strip()
        
        # Security: only allow SELECT
        if not sql.upper().startswith("SELECT"):
            return [TextContent(type="text", 
                text="Error: Only SELECT queries are allowed for safety.")]
        
        try:
            db = get_db()
            cursor = db.execute(sql)
            columns = [desc[0] for desc in cursor.description] if cursor.description else []
            rows = cursor.fetchall()
            db.close()
            
            # Format as markdown table
            if not rows:
                return [TextContent(type="text", text="Query returned no results.")]
            
            header = "| " + " | ".join(columns) + " |"
            separator = "| " + " | ".join("---" for _ in columns) + " |"
            body = "\n".join("| " + " | ".join(str(v) for v in row) + " |" for row in rows[:50])
            
            result = f"{header}\n{separator}\n{body}"
            if len(rows) > 50:
                result += f"\n\n... and {len(rows) - 50} more rows"
            
            return [TextContent(type="text", text=result)]
        
        except Exception as e:
            return [TextContent(type="text", text=f"SQL Error: {e}")]
    
    elif name == "list_tables":
        db = get_db()
        cursor = db.execute(
            "SELECT name, sql FROM sqlite_master WHERE type='table' ORDER BY name"
        )
        tables = cursor.fetchall()
        db.close()
        
        result = "Database tables:\n\n"
        for name, schema in tables:
            result += f"**{name}**\n```sql\n{schema}\n```\n\n"
        return [TextContent(type="text", text=result)]

@app.list_resources()
async def list_resources() -> list[Resource]:
    """Expose each table as a readable resource."""
    db = get_db()
    cursor = db.execute("SELECT name FROM sqlite_master WHERE type='table'")
    tables = cursor.fetchall()
    db.close()
    
    return [
        Resource(
            uri=f"db://tables/{name}",
            name=f"Table: {name}",
            description=f"Contents of the {name} table",
            mimeType="text/plain",
        )
        for (name,) in tables
    ]