my-mcp-server/
├── pyproject.toml
├── server.py
├── tools/
│   ├── __init__.py
│   └── calculator.py
└── tests/
    └── test_tools.py

from mcp.server.fastmcp import FastMCP

mcp = FastMCP("hello-mcp")

@mcp.tool()
def greet(name: str) -> str:
    return f"Hello, {name}! MCP Server is running."

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

{
  "mcpServers": {
    "hello-mcp": {
      "command": "python",
      "args": ["/absolute/path/to/server.py"]
    }
  }
}

from pydantic import Field
import httpx
import sqlite3
from datetime import datetime, timezone

@mcp.tool()
def calculator(expression: str) -> str:
    allowed = set("0123456789+-*/(). ")
    if not all(c in allowed for c in expression):
        raise ValueError("Invalid characters in expression")
    return str(eval(expression))

@mcp.tool()
def read_file(path: str) -> str:
    with open(path, "r", encoding="utf-8") as f:
        return f.read()

@mcp.tool()
def write_file(path: str, content: str) -> str:
    with open(path, "w", encoding="utf-8") as f:
        f.write(content)
    return f"Written {len(content)} chars to {path}"

@mcp.tool()
async def fetch_url(url: str) -> str:
    async with httpx.AsyncClient(timeout=30) as client:
        resp = await client.get(url)
        resp.raise_for_status()
        return resp.text[:8000]

@mcp.tool()
def query_db(sql: str, db_path: str = "data.db") -> str:
    if not sql.strip().upper().startswith("SELECT"):
        raise ValueError("Only SELECT queries allowed")
    conn = sqlite3.connect(db_path)
    rows = conn.execute(sql).fetchall()
    conn.close()
    return str(rows)

@mcp.tool()
def current_time(tz: str = "UTC") -> str:
    return datetime.now(timezone.utc).isoformat()

@mcp.resource("config://app/settings")
def app_settings() -> str:
    return open("settings.json").read()

@mcp.resource("logs://recent/{lines}")
def recent_logs(lines: str) -> str:
    n = min(int(lines), 500)
    with open("/var/log/app.log") as f:
        return "".join(f.readlines()[-n:])

from mcp.types import PromptMessage, TextContent

@mcp.prompt()
def code_review(language: str, focus: str = "security") -> list[PromptMessage]:
    return [
        PromptMessage(role="user", content=TextContent(
            type="text",
            text=f"Review this {language} code focusing on {focus}. "
                 f"Use read_file tool to load the target, then list issues by severity."
        )),
        PromptMessage(role="assistant", content=TextContent(
            type="text",
            text=f"I'll review the {language} code with emphasis on {focus}. "
                 f"Please provide the file path."
        )),
    ]

from mcp.server.fastmcp import FastMCP

mcp = FastMCP("my-server", host="0.0.0.0", port=8080)

if __name__ == "__main__":
    mcp.run(transport="streamable-http")

import pytest
from server import calculator

def test_calculator_basic():
    assert calculator("2 + 3") == "5"

def test_calculator_rejects_injection():
    with pytest.raises(ValueError):
        calculator("__import__('os').system('rm -rf /')")

def test_calculator_division():
    assert calculator("10 / 4") == "2.5"

import chromadb
from watchfiles import awatch
from sentence_transformers import SentenceTransformer

client = chromadb.PersistentClient(path="./chroma_db")
collection = client.get_or_create_collection("knowledge")
embedder = SentenceTransformer("all-MiniLM-L6-v2")

@mcp.tool()
def search_knowledge(query: str, top_k: int = 5) -> str:
    embedding = embedder.encode(query).tolist()
    results = collection.query(query_embeddings=[embedding], n_results=top_k)
    return str(results["documents"])

@mcp.tool()
def index_document(path: str) -> str:
    content = open(path).read()
    doc_id = path.replace("/", "_")
    collection.upsert(
        ids=[doc_id],
        documents=[content],
        embeddings=[embedder.encode(content).tolist()],
    )
    return f"Indexed {path} ({len(content)} chars)"

async def watch_and_reindex(directory: str):
    async for changes in awatch(directory):
        for change in changes:
            if change[1].endswith((".md", ".py", ".ts")):
                index_document(change[1])