Model Context Protocol (MCP): Build AI Tools That Work Everywhere

pip install mcp

# server.py — a simple MCP server with file and database tools
import asyncio
import sqlite3
from pathlib import Path
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp import types

# Create the MCP server
app = Server("my-tools")

# Tool 1: Read a file
@app.list_tools()
async def list_tools() -> list[types.Tool]:
    return [
        types.Tool(
            name="read_file",
            description="Read the contents of a file",
            inputSchema={
                "type": "object",
                "properties": {
                    "path": {
                        "type": "string",
                        "description": "Path to the file to read"
                    }
                },
                "required": ["path"]
       }
        ),
        types.Tool(
            name="query_database",
 local database",
            inputSchema={
                "type": "object",
                "properties": {
                    "query": {
                pe": "string",
                        "description": "SQL SELECT query to execute"
                    }
                },
                "required": ["query"]
  }
        ),
        types.Tool(
            name="list_directory",
on="List files in a directory",
            inputSchema={
bject",
                "properties": {
                 "path": {
                        "type": "string",
                        "description": "Directory path to list"
                    }
                },
                "required": ["path"]
            }
        ),
    ]

@app.call_tool()
async def call_tool(name: str, arguments: dict) -> list[types.TextContent]:
== "read_file":
        path = Path(arguments["path"])
        if not path.exists():
            return [types.Te")]
        if not path.is_file():
            return [types.TextContent(type="text", text=f"Error: Not a file: {path}")]
coding="utf-8")
        return [types.TextContent(type="text", text=content)]

    elif name == "query_database":
        query = arguments["query"]
        # Safety: only allow SELECT queries
"SELECT"):
            return [types.TextContent(type="text", text="Error: Only SELECT queries are allowed")]

        conn = sqlite3.connect("data.db")
    conn.row_factory = sqlite3.Row
        cursor = conn.execute(query)
w) for row in cursor.fetchall()]
        conn.close()

        import json
tContent(type="text", text=json.dumps(rows, indent=2))]

list_directory":
        path = Path(arguments["path"])
ath.exists():
            return [types.TextContent(type="text", text=f"Error: Directory not found: {path}")]

        files = []
 in sorted(path.iterdir()):
            item_type = "dir" if item.is_dir() else "file"
            size = item.stat().st_size if item.is_file() else 0
            files.append(f"{item_type:4} {size:10} {item.name}")

        return [types.TextContent(type="text", text="\n".join(files))]

    else:
ame}")]

# Run the server
async def main():
stream):
        await app.run(read_stream, write_stream, app.create_initialization_options())

if __name__ == "__main__":
())

ntextprotocol/sdk

// server.ts
import { Server } from "@modelcontextprotocol/sdk/server/index.js";

import {
    CallToolRequestSchema,
    ListToolsRequestSchema,
sdk/types.js";
import { readFileSync, readdirSync, statSync } from "fs";
import { join } from "path";

const server = new Server(
    { name: "my-tools", ve1.0.0" },
    { capabilities: { tools: {} } }
);

// Define available tools
sync () => ({
    tools: [
        {
            name: "read_file",
            description: "Read the contents of a file",
            inputSchema: {
                type: "object",
rties: {
                    path: { type: "string", description: "File path to read" }
                },

            }
        },
        {
 "fetch_url",
            description: "Fetch content from a URL",
            inputSchema: {
                type: "object",
          properties: {
                    url: { type: "string", description: "URL to fetch" }
contextprotocol/servers)
.com/modelcontextprotocol/typescript-sdk)
- [MCP Server Registry](https://github.com/modelextprotocol/python-sdk)
- [MCP TypeScript SDK](https://githubdocs/)
- [MCP GitHub](https://github.com/modelcontextprotocol)
- [MCP Python SDK](https://github.com/modelcontocumentation](https://modelcontextprotocol.io/ad_text())]

# Always validate and sanitize tool inputs
@app.call_tool()
async def call_tool(name: str, arguments: dict```

## Security Considerations
"-y", "@modelcontextprotocol/server-postgres", "postgresql://localhost/mydb"]
        }
    }
}
and": "npx",
            "args": [nv": { "GITHUB_PERSONAL_ACCESS_TOKEN": "ghp_..." }
        },
        "postgres": {
            "comm", "@modelcontextprotocol/server-github"],
            "e"github": {
            "command": "npx",
            "args": ["-yl/server-filesystem", "/Users/me"]
        },
        system": {
            "command": "npx",
            "args": ["-y", "@modelcontextprotoco
    "mcpServers": {
        "fileetch URLs

// Claude Desktop config with multiple servers
{lcontextprotocol/server-sqlite /path/to/db.sqlite
npx mcp-server-fetch  # fmunity servers
npx @moder-brave-search  # web search
npx @modelcontextprotocol/server-slack

# Comcalhost/mydb
npx @modelcontextprotocol/serveol/server-github
npx @modelcontextprotocol/server-postgres postgresql://lo /path/to/allow
npx @modelcontextprotoctextprotocol/server-filesystemal servers from Anthropic
npx @modelconcratch — use existing servers:

```bash
# Offici
## Pre-Built MCP Servers

Don't build from sme__ == "__main__":
    asyncio.run(main())

# github_mcp_server.py — let AI interact with GitHub
import asyncio
imp

## Real-World MCP Server: G:", result.content[0].text)

asyncio.run(use_mcp_tools())
```          arguments={"query": "SELECT * FROM users LIMIT 10"}
            )
            print("Query resultstool(
                "query_database",
      ntent[0].text)

            # Query database
            result = await session.call_       )
            print("File contents:", result.co  arguments={"path": "/data/report.csv"}
     .call_tool(
                "read_file",
              ol
            result = await sessionin tools.tools])

            # Call a tols()
            print("Available tools:", [t.name for t           tools = await session.list_too)

            # List available tools
  
            # Initialize
            await session.initialize(ync with ClientSession(read, write) as session:,
    )

    async with stdio_client(server_params) as (read, write):
        ason",
        args=["server.py"]_tools():
    # Connect to your MCP server
    server_params = StdioServerParameters(
        command="pythmcp.client.stdio import stdio_client

async def use_mcp import ClientSession, StdioServerParameters
from  tools in your own application
import asyncio
from mcp

## Using MCP Programmatically

```python
# Use MCPmer icon (🔨) indicating tools are available.rt Claude Desktop. You'll see a hamtocol/server-filesystem", "/Users/me/Documents"]
        }
    }
}

//onnect(transport);