from mcp.server import Server
from mcp.server.models import InitializationOptions
import mcp.types as types

server = Server("my-database-server")

@server.list_tools()
async def handle_list_tools() -> list[types.Tool]:
    return [
        types.Tool(
            name="query_database",
            description="Execute a read-only SQL query against the analytics database",
            inputSchema={
                "type": "object",
                "properties": {
                    "query": {
                        "type": "string",
                        "description": "SQL SELECT query to execute"
                    }
                },
                "required": ["query"]
            }
        )
    ]

@server.call_tool()
async def handle_call_tool(name: str, arguments: dict) -> list[types.TextContent]:
    if name == "query_database":
        result = await db.execute(arguments["query"])
        return [types.TextContent(type="text", text=str(result))]
    raise ValueError(f"Unknown tool: {name}")

#!/usr/bin/env python3
"""MCP server for GitHub integration"""

import asyncio
import httpx
from mcp.server import Server
from mcp.server.stdio import stdio_server
import mcp.types as types

API_TOKEN = os.environ["GITHUB_TOKEN"]
BASE_URL = "https://api.github.com"

server = Server("github-mcp")

async def gh_get(path: str) -> dict:
    async with httpx.AsyncClient() as client:
        r = await client.get(
            f"{BASE_URL}{path}",
            headers={
                "Authorization": f"Bearer {API_TOKEN}",
                "Accept": "application/vnd.github+json",
                "X-GitHub-Api-Version": "2022-11-28",
            }
        )
        r.raise_for_status()
        return r.json()

@server.list_tools()
async def list_tools() -> list[types.Tool]:
    return [
        types.Tool(
            name="list_repos",
            description="List repositories for an organization or user",
            inputSchema={
                "type": "object",
                "properties": {
                    "owner": {"type": "string", "description": "GitHub org or username"}
                },
                "required": ["owner"]
            }
        ),
        types.Tool(
            name="get_pull_requests",
            description="Get open pull requests for a repository",
            inputSchema={
                "type": "object",
                "properties": {
                    "owner": {"type": "string"},
                    "repo": {"type": "string"},
                    "state": {
                        "type": "string",
                        "enum": ["open", "closed", "all"],
                        "default": "open"
                    }
                },
                "required": ["owner", "repo"]
            }
        ),
    ]

@server.call_tool()
async def call_tool(name: str, arguments: dict) -> list[types.TextContent]:
    try:
        if name == "list_repos":
            data = await gh_get(f"/orgs/{arguments['owner']}/repos?per_page=30")
            summary = [{"name": r["name"], "stars": r["stargazers_count"], 
                       "language": r["language"]} for r in data]
            return [types.TextContent(type="text", text=str(summary))]
        
        elif name == "get_pull_requests":
            owner, repo = arguments["owner"], arguments["repo"]
            state = arguments.get("state", "open")
            data = await gh_get(f"/repos/{owner}/{repo}/pulls?state={state}")
            prs = [{"number": pr["number"], "title": pr["title"], 
                   "author": pr["user"]["login"], "created_at": pr["created_at"]} 
                  for pr in data]
            return [types.TextContent(type="text", text=str(prs))]
        
        raise ValueError(f"Unknown tool: {name}")
    
    except httpx.HTTPStatusError as e:
        return [types.TextContent(
            type="text", 
            text=f"GitHub API error: {e.response.status_code} — {e.response.text}"
        )]

async def main():
    async with stdio_server() as (read_stream, write_stream):
        await server.run(read_stream, write_stream, InitializationOptions(
            server_name="github-mcp",
            server_version="0.1.0",
            capabilities=server.get_capabilities(
                notification_options=None,
                experimental_capabilities={}
            )
        ))

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

from mcp.server.fastmcp import FastMCP
from fastapi import FastAPI

# FastMCP makes it trivially easy to expose over HTTP
mcp = FastMCP("company-data-server")

@mcp.tool()
async def query_analytics(sql: str) -> str:
    """Run a read-only query against the analytics warehouse"""
    result = await warehouse.execute(sql)
    return result.to_markdown()

@mcp.resource("metrics://daily-summary")
async def daily_summary() -> str:
    """Today's key business metrics"""
    metrics = await get_daily_metrics()
    return metrics.to_json()

# Mount on FastAPI for HTTP transport
app = FastAPI()
app.mount("/mcp", mcp.get_asgi_app())

import sqlparse

@server.call_tool()
async def call_tool(name: str, arguments: dict):
    if name == "query_database":
        query = arguments["query"]
        # Reject anything that's not a SELECT
        parsed = sqlparse.parse(query)[0]
        if parsed.get_type() != "SELECT":
            raise ValueError("Only SELECT queries are permitted")
        # Use parameterized queries, not string interpolation
        ...