Model Context Protocol training: build secure MCP servers

Model Context Protocol (MCP) is an open standard for connecting AI applications to tools and contextual data. This course teaches the protocol as an engineering system—not as a list of connectors. You will trace the message lifecycle, design safe interfaces, build ten local servers, connect them to major AI clients and finish with a production-ready capstone.

Choose how you want to learn

For every topic, you can watch the concept video, follow a hands-on demonstration or jump directly to the corresponding written documentation below.

You should be comfortable running terminal commands and reading basic Python. Use a disposable project and non-sensitive data while learning. Product interfaces change; the linked official documentation is the source of truth for current configuration.

1. Understand the problem MCP solves

Before MCP, every AI application needed bespoke integrations for every data source and action. MCP separates the AI host from integrations through a shared protocol. A host such as Claude Code, an OpenAI agent or Gemini CLI creates an MCP client, discovers a server's capabilities and calls only the interfaces that server exposes.

MCP does not replace an API. The server normally wraps APIs, databases, files or local programs and presents model-friendly capabilities with descriptions and JSON schemas. The model proposes tool use; the host remains responsible for policy, consent and execution controls.

The three participants

• Host: the AI application coordinating models, permissions and user interaction.
• Client: the protocol component inside the host that maintains a session with one server.
• Server: a program exposing focused capabilities and context.

The official MCP architecture guide describes this one-client-per-server relationship. This boundary matters: a compromised server should not automatically gain access to another server or the host's entire context.

2. Trace the architecture, lifecycle and transports

MCP's data layer uses JSON-RPC 2.0 messages. A session begins with initialize: client and server agree on a protocol version and capabilities, then the client sends an initialized notification. Only then should normal operations begin. Requests have IDs and receive results or errors; notifications do not expect responses.

Choose the transport deliberately

| Transport | Best fit | Important constraint | |---|---|---| | stdio | A trusted local process launched by the host | Never log diagnostics to stdout; it carries protocol messages | | Streamable HTTP | A local service or remote multi-user server | Validate origin, authentication, session handling and timeouts | | HTTP+SSE | Existing legacy servers | Deprecated for new integrations; plan migration |

The current transport specification defines stdio and Streamable HTTP. Transport only moves messages; the data layer defines meaning.

Host                 MCP client                 MCP server
 |--- create ----------->|                           |
 |                       |--- initialize ---------->|
 |                       |<-- version/capabilities--|
 |                       |--- initialized --------->|
 |                       |--- tools/list ---------->|
 |                       |<-- tool definitions -----|
 |--- approve request -->|--- tools/call ---------->|
 |                       |<-- structured result ----|

3. Design tools, resources, prompts and client features

Server features are different interface contracts:

• Tools are model-controlled operations. Use them for calculations, searches and actions. Define narrow input schemas and meaningful errors.
• Resources are application-controlled contextual data identified by URIs. Use them for documents, records and state that can be read without inventing an action.
• Prompts are user-controlled reusable workflow templates. Use them when a person intentionally selects a guided interaction.

Clients may also offer sampling (the server requests model generation), elicitation (the server asks the user for information) and logging. Negotiate these capabilities; never assume they exist.

Bad interface: run_command(command: string). Better interface: get_release_status(environment: "staging" | "production"). The second communicates intent, constrains input and enables meaningful approval.

A tool design checklist

1. Give the tool one job and an action-oriented name.
2. Explain when it should and should not be called.
3. Constrain inputs with enums, lengths, formats and defaults.
4. Return concise structured data plus a human-readable summary.
5. Make read and write operations visibly different.
6. Design idempotency, timeout and failure behaviour.
7. Treat all model-provided arguments as untrusted input.

4. Prepare a local MCP engineering workspace

These builds use Python and the official MCP SDK's FastMCP API. Create one virtual environment, then keep each example in its own file.

mkdir mcp-lab && cd mcp-lab
python -m venv .venv
# macOS/Linux: source .venv/bin/activate
# Windows PowerShell: .venv\Scripts\Activate.ps1
python -m pip install "mcp[cli]" httpx

The smallest server is intentionally boring:

# server.py
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("agileitt-lab")

@mcp.tool()
def ping(message: str = "ready") -> dict[str, str]:
    """Return a health response. This does not access external systems."""
    return {"status": "ok", "echo": message[:100]}

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

Run development inspection with mcp dev server.py, or start it directly with python server.py. Use stderr for diagnostics. Pin dependencies before sharing the project and never place tokens in source control.

5. Build faster with FastMCP

FastMCP is the high-level Python server interface included in the official MCP Python SDK. It turns ordinary typed Python functions into protocol-compliant tools, resources and prompts. It handles message routing, capability advertisement, schema generation and transport plumbing so you can concentrate on the behaviour and security of the integration.

FastMCP is an abstraction over MCP—not a different protocol. Clients still see standard MCP primitives and JSON-RPC messages. This means a FastMCP server can connect to any compatible host, while you retain the option to use the Python SDK's lower-level server API when you need precise protocol control.

Understand what FastMCP generates

For a decorated tool, FastMCP uses:

• the function name as the default tool name;
• the docstring as model-facing guidance;
• Python type hints to generate the input JSON Schema;
• defaults, unions, literals and validation models to constrain arguments; and
• the annotated return value to produce structured output where supported.

That convenience increases the importance of normal Python interface design. A vague docstring or overly broad parameter becomes a vague or overly broad model capability.

from typing import Annotated, Literal
from pydantic import Field
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("release-adviser")

@mcp.tool()
def assess_release(
    environment: Literal["staging", "production"],
    failed_checks: Annotated[int, Field(ge=0, le=100)],
    approved: bool = False,
) -> dict:
    """Assess release readiness. This tool reports advice and never deploys."""
    ready = failed_checks == 0 and (environment == "staging" or approved)
    return {
        "environment": environment,
        "ready": ready,
        "reason": "checks passed" if ready else "failed checks or approval missing",
    }

Inspect the generated schema rather than assuming annotations produced the contract you intended:

mcp dev server.py
# In MCP Inspector: tools/list → assess_release → inspect inputSchema

Use Context for request-scoped MCP capabilities

FastMCP can inject a typed Context into a tool or resource. Context provides request metadata and protocol features such as client-visible logging, progress reporting, resource access, sampling and elicitation when the connected client supports them.

from mcp.server.fastmcp import Context, FastMCP

mcp = FastMCP("import-worker")

@mcp.tool()
async def import_records(count: int, ctx: Context) -> dict:
    """Simulate a bounded import and report progress to the client."""
    if not 1 <= count <= 100:
        raise ValueError("count must be between 1 and 100")
    for current in range(count):
        await ctx.report_progress(current + 1, count)
    await ctx.info(f"Validated {count} training records")
    return {"validated": count, "written": 0}

Context is request-scoped; it should not become an unstructured global dependency. Check client capabilities before designing a workflow that relies on sampling, elicitation or progress UI.

Manage startup and shutdown with lifespan

Use a lifespan function for shared resources such as database pools or HTTP clients. Initialise them once, expose a typed application context and close them reliably when the server stops. Avoid opening a new expensive connection on every tool call.

from collections.abc import AsyncIterator
from contextlib import asynccontextmanager
from dataclasses import dataclass
import httpx
from mcp.server.fastmcp import Context, FastMCP

@dataclass
class AppState:
    http: httpx.AsyncClient

@asynccontextmanager
async def lifespan(server: FastMCP) -> AsyncIterator[AppState]:
    async with httpx.AsyncClient(timeout=8) as client:
        yield AppState(http=client)

mcp = FastMCP("service-reader", lifespan=lifespan)

@mcp.tool()
async def service_health(ctx: Context) -> dict:
    response = await ctx.request_context.lifespan_context.http.get(
        "https://status.example.com/health"
    )
    response.raise_for_status()
    return response.json()

Run locally and remotely

For local clients, mcp.run() uses stdio by default. For a remote deployment, FastMCP can expose Streamable HTTP:

from mcp.server.fastmcp import FastMCP

mcp = FastMCP(
    "remote-training-server",
    stateless_http=True,
    json_response=True,
)

@mcp.tool()
def health() -> dict[str, str]:
    return {"status": "ok"}

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

Stateless HTTP can simplify horizontal scaling, but it does not supply authentication, tenant isolation, rate limiting or safe tool design automatically. Add those controls at the application and infrastructure layers. Use the official MCP Python SDK server guide for current FastMCP APIs and deployment options.

FastMCP or the low-level server API?

| Choose FastMCP when | Consider the low-level API when | |---|---| | Python decorators fit your component model | You need custom protocol handlers | | Generated schemas match the intended contract | You need exact message-level control | | You want rapid tools/resources/prompts development | You are implementing experimental protocol behaviour | | Standard transports and lifecycle are sufficient | Your server has unusual routing or negotiation requirements |

6. Build local servers 1–3: tools, resources and prompts

Each example introduces one concept. You may combine them later, but first make every boundary observable and testable.

Build 1 — typed calculator tool

from typing import Literal
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("calculator")

@mcp.tool()
def calculate(operation: Literal["add", "subtract", "multiply", "divide"],
              left: float, right: float) -> dict:
    """Perform one basic arithmetic operation."""
    if operation == "divide" and right == 0:
        raise ValueError("right must be non-zero for division")
    result = {"add": left + right, "subtract": left - right,
              "multiply": left * right,
              "divide": left / right if right else 0}[operation]
    return {"operation": operation, "result": result}

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

Build 2 — addressable notes resources

from mcp.server.fastmcp import FastMCP

mcp = FastMCP("notes")
NOTES = {"welcome": "MCP keeps integrations separate from the host.",
         "safety": "Treat tool inputs and remote content as untrusted."}

@mcp.resource("notes://{name}")
def read_note(name: str) -> str:
    """Read a named training note."""
    if name not in NOTES:
        raise ValueError("unknown note")
    return NOTES[name]

@mcp.resource("notes://index")
def note_index() -> str:
    return "\n".join(sorted(NOTES))

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

Build 3 — reusable review prompt

from mcp.server.fastmcp import FastMCP

mcp = FastMCP("review-prompts")

@mcp.prompt()
def review_change(goal: str, diff: str) -> str:
    """Create a disciplined software-change review workflow."""
    return f"""Review this change against its goal.
Goal: {goal}
Diff: {diff}
Check correctness, security, accessibility and missing tests.
Separate evidence from assumptions. Do not execute the diff."""

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

7. Build local servers 4–6: files, SQLite and HTTP APIs

Real connectors touch systems with risk. Scope access before adding capability.

Build 4 — workspace-bound file reader

from pathlib import Path
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("safe-files")
ROOT = Path("./sample-files").resolve()
ALLOWED = {".md", ".txt", ".json"}

def safe_path(relative_path: str) -> Path:
    candidate = (ROOT / relative_path).resolve()
    if candidate != ROOT and ROOT not in candidate.parents:
        raise ValueError("path escapes the allowed workspace")
    if candidate.suffix.lower() not in ALLOWED:
        raise ValueError("file type is not allowed")
    return candidate

@mcp.tool()
def read_workspace_file(relative_path: str) -> dict:
    """Read one text file below the configured sample-files directory."""
    path = safe_path(relative_path)
    return {"path": str(path.relative_to(ROOT)),
            "content": path.read_text(encoding="utf-8")[:50_000]}

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

Build 5 — read-only SQLite reporting

import sqlite3
from typing import Literal
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("sales-report")
DB_URI = "file:training.db?mode=ro"

@mcp.tool()
def sales_summary(period: Literal["week", "month"], limit: int = 10) -> list[dict]:
    """Return top products from the read-only training database."""
    days = 7 if period == "week" else 30
    limit = max(1, min(limit, 50))
    sql = """SELECT product, ROUND(SUM(amount), 2) total
             FROM sales WHERE sold_at >= date('now', ?)
             GROUP BY product ORDER BY total DESC LIMIT ?"""
    with sqlite3.connect(DB_URI, uri=True) as connection:
        rows = connection.execute(sql, (f"-{days} days", limit)).fetchall()
    return [{"product": row[0], "total": row[1]} for row in rows]

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

Do not expose a generic SQL executor. Parameterised queries prevent injection, while a read-only connection and allowlisted report shape limit impact.

Build 6 — timeout-bound weather API

import httpx
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("weather")

@mcp.tool()
async def current_weather(latitude: float, longitude: float) -> dict:
    """Read current public weather for valid coordinates."""
    if not (-90 <= latitude <= 90 and -180 <= longitude <= 180):
        raise ValueError("invalid coordinates")
    async with httpx.AsyncClient(timeout=8) as client:
        response = await client.get("https://api.open-meteo.com/v1/forecast",
            params={"latitude": latitude, "longitude": longitude,
                    "current": "temperature_2m,wind_speed_10m"})
        response.raise_for_status()
        return response.json()["current"]

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

8. Build local servers 7–10: Git, CSV, docs and tasks

Build 7 — safe Git inspection

import subprocess
from pathlib import Path
from typing import Literal
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("git-reader")
REPO = Path("./demo-repo").resolve()

@mcp.tool()
def inspect_git(view: Literal["status", "recent-log", "diff-stat"]) -> str:
    """Inspect an allowlisted Git view without changing the repository."""
    args = {"status": ["status", "--short"],
            "recent-log": ["log", "-10", "--oneline"],
            "diff-stat": ["diff", "--stat"]}[view]
    result = subprocess.run(["git", *args], cwd=REPO, capture_output=True,
                            text=True, timeout=10, check=True)
    return result.stdout[:20_000]

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

Never concatenate model text into a shell command. Pass a fixed executable and allowlisted argument arrays.

Build 8 — CSV profile tool

import csv
from pathlib import Path
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("csv-profile")
ROOT = Path("./data").resolve()

@mcp.tool()
def profile_csv(filename: str) -> dict:
    """Return columns, row count and missing-value counts for one CSV file."""
    path = (ROOT / filename).resolve()
    if ROOT not in path.parents or path.suffix.lower() != ".csv":
        raise ValueError("only CSV files below data are allowed")
    with path.open(newline="", encoding="utf-8") as handle:
        rows = list(csv.DictReader(handle))
    columns = list(rows[0]) if rows else []
    return {"rows": len(rows), "columns": columns,
            "missing": {c: sum(not r.get(c) for r in rows) for c in columns}}

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

Build 9 — local documentation search

from pathlib import Path
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("docs-search")
DOCS = Path("./docs").resolve()

@mcp.tool()
def search_docs(query: str, limit: int = 5) -> list[dict]:
    """Search local Markdown headings and lines using case-insensitive terms."""
    terms = [term.lower() for term in query.split() if len(term) > 2][:8]
    if not terms: raise ValueError("provide a meaningful query")
    matches = []
    for path in DOCS.rglob("*.md"):
        for number, line in enumerate(path.read_text(encoding="utf-8").splitlines(), 1):
            score = sum(term in line.lower() for term in terms)
            if score: matches.append({"score": score, "file": str(path.relative_to(DOCS)),
                                      "line": number, "text": line[:300]})
    return sorted(matches, key=lambda item: item["score"], reverse=True)[:min(limit, 20)]

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

Build 10 — approval-friendly task creation

import json, secrets
from pathlib import Path
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("task-board")
STORE = Path("./tasks.json")

@mcp.tool()
def preview_task(title: str, description: str) -> dict:
    """Preview a task without saving it."""
    return {"title": title.strip()[:120], "description": description.strip()[:2000],
            "will_write": str(STORE.resolve())}

@mcp.tool()
def create_task(title: str, description: str, approved: bool = False) -> dict:
    """Create a local task only after explicit approval."""
    if not approved: raise ValueError("preview and obtain user approval first")
    tasks = json.loads(STORE.read_text()) if STORE.exists() else []
    task = {"id": secrets.token_hex(4), "title": title.strip()[:120],
            "description": description.strip()[:2000], "status": "open"}
    tasks.append(task)
    STORE.write_text(json.dumps(tasks, indent=2), encoding="utf-8")
    return task

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

The preview/commit split makes consequence visible before mutation. In production, use host-level approval too; a boolean argument alone is not proof of human consent.

9. Connect local MCP servers to Claude

Claude Code can register local stdio servers from the command line. From the project containing server.py:

claude mcp add --transport stdio mcp-lab -- python server.py
claude mcp list
claude mcp get mcp-lab

Use project scope when teammates should share configuration and user scope only for trusted personal utilities. Project-scoped configuration is stored in .mcp.json; review it like executable code before committing. On native Windows, package commands may need cmd /c, for example claude mcp add --transport stdio demo -- cmd /c npx -y package-name.

Claude Desktop also launches local servers from its MCP configuration. Use an absolute command/path, restart the app after changes and inspect its logs if the server is not discovered. Keep credentials in environment variables, not JSON checked into Git.

Verify before trusting

1. List registered servers and confirm the command.
2. Ask Claude which tools are available.
3. Trigger a read-only call first.
4. Inspect arguments before approving a write.
5. Confirm the actual external state after the response.

Follow the current Claude Code MCP documentation for scopes, OAuth and management commands.

10. Connect MCP servers to OpenAI

OpenAI's Agents SDK supports two distinct patterns. For a local subprocess, your application owns an MCPServerStdio connection:

import asyncio
from agents import Agent, Runner
from agents.mcp import MCPServerStdio

async def main():
    async with MCPServerStdio(
        name="Agileitt MCP lab",
        params={"command": "python", "args": ["server.py"]},
    ) as server:
        agent = Agent(name="MCP learner", mcp_servers=[server],
                      instructions="Use tools only when needed; explain consequential calls.")
        result = await Runner.run(agent, "Use the calculator to multiply 17 by 24.")
        print(result.final_output)

asyncio.run(main())

For a publicly reachable server, a hosted MCP tool lets the Responses platform manage the remote round trip. Require approval for consequential tools and expose the smallest useful tool set.

from agents import Agent, HostedMCPTool

agent = Agent(
    name="Remote MCP learner",
    tools=[HostedMCPTool(tool_config={
        "type": "mcp",
        "server_label": "company_docs",
        "server_url": "https://mcp.example.com/mcp",
        "require_approval": "always"
    })]
)

The official OpenAI Agents SDK MCP guide also documents Streamable HTTP, tool filtering, caching, tracing and approval policies. Never send a local-only URL to a hosted integration; OpenAI's infrastructure must be able to reach a hosted server.

11. Connect MCP servers to Gemini

Gemini CLI reads MCP definitions from settings.json. A local server is configured with command and args:

{
  "mcpServers": {
    "agileitt-lab": {
      "command": "python",
      "args": ["server.py"],
      "cwd": "/absolute/path/to/mcp-lab",
      "timeout": 30000,
      "trust": false,
      "includeTools": ["calculate", "search_docs"]
    }
  }
}

Use gemini mcp list or /mcp list to diagnose discovery. includeTools creates an allowlist, while trust: false preserves confirmation. Avoid underscores in server names because current policy parsing uses the generated MCP tool namespace.

A Streamable HTTP server uses httpUrl:

{
  "mcpServers": {
    "team-docs": {
      "httpUrl": "https://mcp.example.com/mcp",
      "headers": {"Authorization": "Bearer $TEAM_MCP_TOKEN"},
      "includeTools": ["search_docs"]
    }
  }
}

The current Gemini CLI MCP server guide covers server discovery, namespaces, OAuth and filtering. Gemini API and agent products have their own integration surfaces, so do not assume CLI configuration automatically applies to every Gemini application.

12. Evaluate and connect remote MCP servers

Remote MCP changes the trust boundary: code executes outside your machine, data crosses a network and authentication must be designed for multiple users. Prefer Streamable HTTP, HTTPS and OAuth where supported. The MCP authorisation specification forbids token passthrough; a server must validate tokens intended for itself.

Remote example 1 — GitHub

GitHub's official MCP server can run locally in Docker or connect remotely where supported. Start with read-only repository tools, a fine-grained token and the smallest repository scope. Review the GitHub MCP server repository for current installation and toolsets.

Remote example 2 — Sentry

Sentry's remote MCP integration can expose issue and project context to supported hosts. Use OAuth, restrict organisation access and begin with investigation workflows before enabling mutations. Check Sentry's MCP documentation for its current endpoint and client setup.

Remote example 3 — Cloudflare

Cloudflare publishes remote MCP servers for parts of its platform. Select one server rather than granting blanket account access, inspect every advertised tool and use a non-production account while learning. See Cloudflare's remote MCP server documentation.

Remote example 4 — your own Streamable HTTP service

When deploying Build 10, replace local JSON storage with authenticated tenant-aware storage, add HTTPS, OAuth resource metadata, per-user authorisation, rate limits, audit events and idempotency keys. Bind local development servers to loopback and validate the HTTP Origin header to reduce DNS-rebinding risk.

Before connecting any remote server, answer:

• Who operates it, and how will you verify the endpoint?
• What data leaves the host and where is it retained?
• Which tools are read-only, destructive or financially consequential?
• Can tools be allowlisted and require approval?
• How are access revoked, tokens rotated and calls audited?
• What happens when tool descriptions or schemas change?

13. Secure, test and operate MCP in production

MCP servers sit at a high-value boundary between model output and real systems. Prompt injection can arrive through user input, retrieved resources or tool results. A trusted server can still return untrusted content. Build controls outside the prompt.

Production control stack

| Layer | Controls | |---|---| | Identity | OAuth audience validation, short-lived credentials, per-user identity | | Authorisation | Least privilege, tenant checks, read/write separation, tool allowlists | | Input | Schema constraints, canonical paths, parameterised queries, size limits | | Execution | Sandboxing, timeouts, egress rules, idempotency, human approval | | Output | Secret/PII filtering, bounded results, provenance and structured errors | | Operations | Audit logs, metrics, tracing, version pinning and incident revocation |

Test protocol behaviour as well as business logic:

1. Initialise with supported and unsupported protocol versions.
2. Snapshot tool/resource/prompt schemas to detect breaking changes.
3. Fuzz paths, URLs, oversized values and Unicode edge cases.
4. Simulate dependency timeouts, partial failures and disconnects.
5. Verify tenant isolation and least-privilege credentials.
6. Confirm write operations need approval and can be retried safely.
7. Redact secrets from logs and traces.
8. Re-run the suite against every supported host.

Capstone: a production-ready project assistant

Combine the safe file reader, documentation search, Git inspection and task preview into one focused server. Add a read-only default role, explicit task-creation approval, structured results, tests, an Inspector evidence pack and configurations for Claude, OpenAI and Gemini. Then expose it through Streamable HTTP in a disposable environment and complete the remote-server review checklist.

Your definition of done is not “the model called a tool”. It is: the right capability was discovered, inputs were validated, authority was enforced, results were traceable, failures were safe and a human could understand the consequence.

Continue with primary sources

• MCP specification
• Build an MCP server
• MCP architecture
• Claude Code MCP
• OpenAI Agents SDK MCP
• Gemini CLI MCP servers