from fastmcp import FastMCP

mcp = FastMCP("hello-world")

@mcp.tool
def greet(name: str) -> str:
    """Say hello to someone."""
    return f"Hello, {name}!"

if __name__ == "__main__":
    mcp.run()

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

const server = new Server(
  { name: "hello-world", version: "1.0.0" },
  { capabilities: { tools: {} } }
);

server.setRequestHandler(ListToolsRequestSchema, async () => ({
  tools: [{
    name: "greet",
    description: "Say hello to someone",
    inputSchema: { type: "object", properties: { name: { type: "string" } } }
  }]
}));

const transport = new StdioServerTransport();
await server.connect(transport);

from fastmcp import FastMCP
import asyncpg

mcp = FastMCP("postgres-mcp")
DATABASE_URL = "postgresql://user:pass@localhost/mydb"

@mcp.tool
async def query(sql: str) -> list[dict]:
    """Execute a read-only SQL query on PostgreSQL."""
    conn = await asyncpg.connect(DATABASE_URL)
    try:
        rows = await conn.fetch(sql)
        return [dict(row) for row in rows]
    finally:
        await conn.close()

@mcp.tool
async def list_tables() -> list[str]:
    """List all tables in the database."""
    sql = """
    SELECT table_name FROM information_schema.tables
    WHERE table_schema = 'public'
    """
    result = await query(sql)
    return [row['table_name'] for row in result]

@mcp.tool
async def describe_table(table_name: str) -> list[dict]:
    """Get column info for a specific table."""
    sql = f"""
    SELECT column_name, data_type, is_nullable
    FROM information_schema.columns
    WHERE table_name = '{table_name}'
    """
    return await query(sql)

import { Server } from "@modelcontextprotocol/sdk/server/index.js";
import Database from "better-sqlite3";

const db = new Database("./data.sqlite");

const server = new Server(
  { name: "sqlite-mcp", version: "1.0.0" },
  { capabilities: { tools: {} } }
);

server.setRequestHandler(CallToolRequestSchema, async (request) => {
  const { name, arguments: args } = request.params;

  if (name === "query") {
    const { sql } = args as { sql: string };
    const result = db.prepare(sql).all();
    return {
      content: [{ type: "text", text: JSON.stringify(result, null, 2) }]
    };
  }

  if (name === "execute") {
    const { sql } = args as { sql: string };
    const result = db.prepare(sql).run();
    return {
      content: [{ type: "text", text: `Affected rows: ${result.changes}` }]
    };
  }
});

from fastmcp import FastMCP
from pathlib import Path
import os

mcp = FastMCP("filesystem-mcp")

@mcp.tool
def read_file(path: str) -> str:
    """Read the contents of a file."""
    file_path = Path(path).expanduser()
    if not file_path.exists():
        raise FileNotFoundError(f"File not found: {path}")
    return file_path.read_text()

@mcp.tool
def write_file(path: str, content: str) -> str:
    """Write content to a file. Creates directories if needed."""
    file_path = Path(path).expanduser()
    file_path.parent.mkdir(parents=True, exist_ok=True)
    file_path.write_text(content)
    return f"Written {len(content)} bytes to {path}"

@mcp.tool
def list_directory(path: str = ".") -> list[dict]:
    """List files and directories at a path."""
    dir_path = Path(path).expanduser()
    items = []
    for item in dir_path.iterdir():
        items.append({
            "name": item.name,
            "type": "directory" if item.is_dir() else "file",
            "size": item.stat().st_size if item.is_file() else None
        })
    return sorted(items, key=lambda x: (x["type"] == "file", x["name"]))

@mcp.tool
def search_files(pattern: str, directory: str = ".") -> list[str]:
    """Search for files matching a glob pattern."""
    dir_path = Path(directory).expanduser()
    return [str(p) for p in dir_path.rglob(pattern)]

import { Server } from "@modelcontextprotocol/sdk/server/index.js";
import { watch, FSWatcher } from "fs";
import { readdir, stat } from "fs/promises";
import { join } from "path";

const server = new Server(
  { name: "file-watcher-mcp", version: "1.0.0" },
  { capabilities: { tools: {} } }
);

let watchers: Map<string, FSWatcher> = new Map();

server.setRequestHandler(CallToolRequestSchema, async (request) => {
  const { name, arguments: args } = request.params;

  if (name === "watch_directory") {
    const { path } = args as { path: string };

    if (watchers.has(path)) {
      return { content: [{ type: "text", text: `Already watching ${path}` }] };
    }

    const watcher = watch(path, { recursive: true }, (event, filename) => {
      console.error(`[${event}] ${filename}`);
    });

    watchers.set(path, watcher);
    return { content: [{ type: "text", text: `Now watching ${path}` }] };
  }

  if (name === "stop_watching") {
    const { path } = args as { path: string };
    const watcher = watchers.get(path);
    if (watcher) {
      watcher.close();
      watchers.delete(path);
    }
    return { content: [{ type: "text", text: `Stopped watching ${path}` }] };
  }
});

from fastmcp import FastMCP
import httpx
from typing import Optional

mcp = FastMCP("rest-api-mcp")

BASE_URL = "https://api.example.com/v1"
API_KEY = os.environ.get("API_KEY")

@mcp.tool
async def api_get(endpoint: str, params: Optional[dict] = None) -> dict:
    """Make a GET request to the API."""
    async with httpx.AsyncClient() as client:
        response = await client.get(
            f"{BASE_URL}{endpoint}",
            params=params,
            headers={"Authorization": f"Bearer {API_KEY}"}
        )
        response.raise_for_status()
        return response.json()

@mcp.tool
async def api_post(endpoint: str, data: dict) -> dict:
    """Make a POST request to the API."""
    async with httpx.AsyncClient() as client:
        response = await client.post(
            f"{BASE_URL}{endpoint}",
            json=data,
            headers={"Authorization": f"Bearer {API_KEY}"}
        )
        response.raise_for_status()
        return response.json()

@mcp.tool
async def api_list_endpoints() -> list[str]:
    """List available API endpoints."""
    return [
        "/users - User management",
        "/products - Product catalog",
        "/orders - Order processing",
        "/analytics - Usage statistics"
    ]

import { Server } from "@modelcontextprotocol/sdk/server/index.js";
import { OAuth2Client } from "google-auth-library";

const oauth2Client = new OAuth2Client({
  clientId: process.env.GOOGLE_CLIENT_ID,
  clientSecret: process.env.GOOGLE_CLIENT_SECRET,
  redirectUri: "http://localhost:3000/callback"
});

const server = new Server(
  { name: "oauth-api-mcp", version: "1.0.0" },
  { capabilities: { tools: {} } }
);

server.setRequestHandler(CallToolRequestSchema, async (request) => {
  const { name, arguments: args } = request.params;

  if (name === "get_auth_url") {
    const url = oauth2Client.generateAuthUrl({
      access_type: "offline",
      scope: ["https://www.googleapis.com/auth/calendar.readonly"]
    });
    return { content: [{ type: "text", text: url }] };
  }

  if (name === "exchange_code") {
    const { code } = args as { code: string };
    const { tokens } = await oauth2Client.getToken(code);
    oauth2Client.setCredentials(tokens);
    return { content: [{ type: "text", text: "Authenticated successfully" }] };
  }

  if (name === "calendar_events") {
    const response = await oauth2Client.request({
      url: "https://www.googleapis.com/calendar/v3/calendars/primary/events"
    });
    return { content: [{ type: "text", text: JSON.stringify(response.data) }] };
  }
});

from fastmcp import FastMCP
import subprocess
from typing import Optional

mcp = FastMCP("git-mcp")

def run_git(args: list[str], cwd: Optional[str] = None) -> str:
    """Run a git command and return output."""
    result = subprocess.run(
        ["git"] + args,
        capture_output=True,
        text=True,
        cwd=cwd
    )
    if result.returncode != 0:
        raise Exception(f"Git error: {result.stderr}")
    return result.stdout

@mcp.tool
def git_status(repo_path: str = ".") -> str:
    """Get current git status."""
    return run_git(["status", "--porcelain"], repo_path)

@mcp.tool
def git_diff(file: Optional[str] = None, repo_path: str = ".") -> str:
    """Show git diff for a file or all changes."""
    args = ["diff"]
    if file:
        args.append(file)
    return run_git(args, repo_path)

@mcp.tool
def git_log(count: int = 10, repo_path: str = ".") -> str:
    """Show recent commit history."""
    return run_git(
        ["log", f"-{count}", "--oneline", "--decorate"],
        repo_path
    )

@mcp.tool
def git_branch_list(repo_path: str = ".") -> str:
    """List all branches."""
    return run_git(["branch", "-a"], repo_path)

@mcp.tool
def git_stash_changes(message: str, repo_path: str = ".") -> str:
    """Stash current changes with a message."""
    return run_git(["stash", "push", "-m", message], repo_path)

import { Server } from "@modelcontextprotocol/sdk/server/index.js";
import { exec } from "child_process";
import { promisify } from "util";

const execAsync = promisify(exec);

const server = new Server(
  { name: "docker-mcp", version: "1.0.0" },
  { capabilities: { tools: {} } }
);

async function docker(args: string): Promise<string> {
  const { stdout, stderr } = await execAsync(`docker ${args}`);
  if (stderr && !stdout) throw new Error(stderr);
  return stdout;
}

server.setRequestHandler(CallToolRequestSchema, async (request) => {
  const { name, arguments: args } = request.params;

  switch (name) {
    case "list_containers":
      const containers = await docker("ps --format '{{.Names}}\t{{.Status}}'");
      return { content: [{ type: "text", text: containers }] };

    case "container_logs":
      const { container, lines = 50 } = args as { container: string; lines?: number };
      const logs = await docker(`logs --tail ${lines} ${container}`);
      return { content: [{ type: "text", text: logs }] };

    case "restart_container":
      const { name: containerName } = args as { name: string };
      await docker(`restart ${containerName}`);
      return { content: [{ type: "text", text: `Restarted ${containerName}` }] };
  }
});

from fastmcp import FastMCP
from slack_sdk import WebClient
from slack_sdk.errors import SlackApiError
import os

mcp = FastMCP("slack-mcp")
client = WebClient(token=os.environ["SLACK_BOT_TOKEN"])

@mcp.tool
def send_message(channel: str, text: str) -> dict:
    """Send a message to a Slack channel."""
    try:
        response = client.chat_postMessage(channel=channel, text=text)
        return {
            "success": True,
            "timestamp": response["ts"],
            "channel": response["channel"]
        }
    except SlackApiError as e:
        return {"success": False, "error": str(e)}

@mcp.tool
def list_channels() -> list[dict]:
    """List all accessible Slack channels."""
    response = client.conversations_list(types="public_channel,private_channel")
    return [
        {"id": c["id"], "name": c["name"], "is_private": c["is_private"]}
        for c in response["channels"]
    ]

@mcp.tool
def get_channel_history(channel: str, limit: int = 20) -> list[dict]:
    """Get recent messages from a channel."""
    response = client.conversations_history(channel=channel, limit=limit)
    return [
        {"user": m.get("user"), "text": m.get("text"), "ts": m["ts"]}
        for m in response["messages"]
    ]

@mcp.tool
def add_reaction(channel: str, timestamp: str, emoji: str) -> dict:
    """Add a reaction to a message."""
    try:
        client.reactions_add(channel=channel, timestamp=timestamp, name=emoji)
        return {"success": True}
    except SlackApiError as e:
        return {"success": False, "error": str(e)}

import { Server } from "@modelcontextprotocol/sdk/server/index.js";
import { Client } from "@notionhq/client";

const notion = new Client({ auth: process.env.NOTION_TOKEN });

const server = new Server(
  { name: "notion-mcp", version: "1.0.0" },
  { capabilities: { tools: {} } }
);

server.setRequestHandler(CallToolRequestSchema, async (request) => {
  const { name, arguments: args } = request.params;

  if (name === "search") {
    const { query } = args as { query: string };
    const response = await notion.search({
      query,
      filter: { property: "object", value: "page" }
    });
    return {
      content: [{
        type: "text",
        text: JSON.stringify(response.results.map(page => ({
          id: page.id,
          title: page.properties?.title?.title?.[0]?.plain_text || "Untitled"
        })))
      }]
    };
  }

  if (name === "get_page") {
    const { page_id } = args as { page_id: string };
    const page = await notion.pages.retrieve({ page_id });
    const blocks = await notion.blocks.children.list({ block_id: page_id });
    return {
      content: [{
        type: "text",
        text: JSON.stringify({ page, blocks: blocks.results })
      }]
    };
  }

  if (name === "create_page") {
    const { parent_id, title, content } = args as {
      parent_id: string;
      title: string;
      content: string;
    };
    const page = await notion.pages.create({
      parent: { page_id: parent_id },
      properties: {
        title: { title: [{ text: { content: title } }] }
      },
      children: [{
        object: "block",
        type: "paragraph",
        paragraph: { rich_text: [{ text: { content } }] }
      }]
    });
    return { content: [{ type: "text", text: `Created page: ${page.id}` }] };
  }
});

from fastmcp import FastMCP
import chromadb
from chromadb.config import Settings
import uuid

mcp = FastMCP("memory-mcp")

client = chromadb.Client(Settings(
    chroma_db_impl="duckdb+parquet",
    persist_directory="./memory_store"
))

collection = client.get_or_create_collection(
    name="memories",
    metadata={"hnsw:space": "cosine"}
)

@mcp.tool
def remember(content: str, tags: list[str] = None) -> dict:
    """Store a memory for later retrieval."""
    memory_id = str(uuid.uuid4())
    metadata = {"tags": ",".join(tags)} if tags else {}

    collection.add(
        documents=[content],
        ids=[memory_id],
        metadatas=[metadata]
    )

    return {"id": memory_id, "status": "remembered"}

@mcp.tool
def recall(query: str, limit: int = 5) -> list[dict]:
    """Search memories by semantic similarity."""
    results = collection.query(
        query_texts=[query],
        n_results=limit
    )

    memories = []
    for i, doc in enumerate(results["documents"][0]):
        memories.append({
            "content": doc,
            "id": results["ids"][0][i],
            "distance": results["distances"][0][i] if results.get("distances") else None
        })

    return memories

@mcp.tool
def forget(memory_id: str) -> dict:
    """Delete a specific memory."""
    collection.delete(ids=[memory_id])
    return {"status": "forgotten", "id": memory_id}

@mcp.tool
def list_memories(limit: int = 20) -> list[dict]:
    """List recent memories."""
    results = collection.peek(limit=limit)
    return [
        {"id": id, "content": doc[:100] + "..."}
        for id, doc in zip(results["ids"], results["documents"])
    ]

from fastmcp import FastMCP
from typing import Optional
import logging

# Setup
mcp = FastMCP("production-server")
logger = logging.getLogger(__name__)

# Tools for actions
@mcp.tool
def create_item(name: str, data: dict) -> dict:
    """Create a new item with the given data."""
    try:
        # Your creation logic here
        item_id = save_to_database(name, data)
        return {"success": True, "id": item_id}
    except Exception as e:
        logger.error(f"Failed to create item: {e}")
        return {"success": False, "error": str(e)}

@mcp.tool
def get_item(item_id: str) -> Optional[dict]:
    """Retrieve an item by ID."""
    return fetch_from_database(item_id)

@mcp.tool
def update_item(item_id: str, updates: dict) -> dict:
    """Update an existing item."""
    try:
        update_in_database(item_id, updates)
        return {"success": True}
    except Exception as e:
        return {"success": False, "error": str(e)}

@mcp.tool
def delete_item(item_id: str) -> dict:
    """Delete an item by ID."""
    try:
        delete_from_database(item_id)
        return {"success": True}
    except Exception as e:
        return {"success": False, "error": str(e)}

# Resources for data access
@mcp.resource("items://{item_id}")
def item_resource(item_id: str) -> dict:
    """Resource for accessing item data directly."""
    return fetch_from_database(item_id)

@mcp.resource("items://list")
def items_list_resource() -> list[dict]:
    """Resource for listing all items."""
    return list_all_items()

# Prompts for common workflows
@mcp.prompt
def create_workflow() -> str:
    """Prompt for creating new items."""
    return """
    To create a new item, I need:
    1. A name for the item
    2. The data to store (as a dictionary)

    Example: create_item(name="test", data={"key": "value"})
    """

import { Server } from "@modelcontextprotocol/sdk/server/index.js";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
import { z } from "zod";

// Schema definitions
const CreateItemSchema = z.object({
  name: z.string().min(1),
  data: z.record(z.unknown())
});

const UpdateItemSchema = z.object({
  id: z.string().uuid(),
  updates: z.record(z.unknown())
});

// Server setup
const server = new Server(
  { name: "production-server", version: "1.0.0" },
  { capabilities: { tools: {}, resources: {} } }
);

// Tool definitions
server.setRequestHandler(ListToolsRequestSchema, async () => ({
  tools: [
    {
      name: "create_item",
      description: "Create a new item",
      inputSchema: {
        type: "object",
        properties: {
          name: { type: "string" },
          data: { type: "object" }
        },
        required: ["name", "data"]
      }
    },
    {
      name: "get_item",
      description: "Get an item by ID",
      inputSchema: {
        type: "object",
        properties: {
          id: { type: "string" }
        },
        required: ["id"]
      }
    },
    {
      name: "update_item",
      description: "Update an existing item",
      inputSchema: {
        type: "object",
        properties: {
          id: { type: "string" },
          updates: { type: "object" }
        },
        required: ["id", "updates"]
      }
    },
    {
      name: "delete_item",
      description: "Delete an item",
      inputSchema: {
        type: "object",
        properties: {
          id: { type: "string" }
        },
        required: ["id"]
      }
    }
  ]
}));

// Tool handlers
server.setRequestHandler(CallToolRequestSchema, async (request) => {
  const { name, arguments: args } = request.params;

  try {
    switch (name) {
      case "create_item": {
        const { name, data } = CreateItemSchema.parse(args);
        const id = await saveToDatabase(name, data);
        return { content: [{ type: "text", text: JSON.stringify({ success: true, id }) }] };
      }

      case "get_item": {
        const { id } = args as { id: string };
        const item = await fetchFromDatabase(id);
        return { content: [{ type: "text", text: JSON.stringify(item) }] };
      }

      case "update_item": {
        const { id, updates } = UpdateItemSchema.parse(args);
        await updateInDatabase(id, updates);
        return { content: [{ type: "text", text: JSON.stringify({ success: true }) }] };
      }

      case "delete_item": {
        const { id } = args as { id: string };
        await deleteFromDatabase(id);
        return { content: [{ type: "text", text: JSON.stringify({ success: true }) }] };
      }

      default:
        throw new Error(`Unknown tool: ${name}`);
    }
  } catch (error) {
    return {
      content: [{
        type: "text",
        text: JSON.stringify({ success: false, error: String(error) })
      }]
    };
  }
});

// Start server
const transport = new StdioServerTransport();
await server.connect(transport);

npx @modelcontextprotocol/inspector your-server-command