3. MCP Servers: Exposing Tools

Overview

In the previous capsule you understood MCP's architecture: host, client, server, transport. All of that was necessary theory. Now you're going to get your hands dirty. This capsule is 100% practical: by the end of it you'll have an MCP server running on your machine that exposes tools, resources, and prompts that any MCP client can discover and use.

The core idea: an MCP server is a Python program that declares which tools it offers and exposes them through a standard protocol. The official SDK includes FastMCP, a high-level interface that lets you declare tools with decorators — if you've used FastAPI, the experience is almost identical. You write Python functions, decorate them, and the framework handles the protocol, the serialization, and the transport.

Why do we start with the server and not the client? Because without a server there's nothing to consume. And because building the server gives you a deep understanding of what an agent "sees" when it connects: tool names, schemas, descriptions, available resources.

SDK Setup

pip install mcp

This installs the mcp package, which includes:

• mcp.server.fastmcp.FastMCP — The high-level interface for creating servers
• mcp.server.Server — The low-level interface (more control, more code)
• mcp.client — Tools for creating clients (capsule 04)
• mcp dev — A CLI command for local testing

Verify the installation:

python -c "from mcp.server.fastmcp import FastMCP; print('MCP SDK OK')"

For a typical MCP server you only need one import:

from mcp.server.fastmcp import FastMCP

FastMCP is all you need to declare tools, resources, prompts and launch the server. Only drop down to the low-level Server when you need granular control over the protocol.

If your tools make HTTP calls, also install httpx:

pip install httpx

Creating a Basic MCP Server

Your first server in 15 lines

from mcp.server.fastmcp import FastMCP

mcp = FastMCP("demo-server")

@mcp.tool()
def add(a: int, b: int) -> int:
    """Add two integers."""
    return a + b

@mcp.tool()
def greet(name: str) -> str:
    """Greet a person by name."""
    return f"Hello, {name}! Welcome."

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

Save this as demo_server.py. The server sits listening on stdin/stdout waiting for JSON-RPC messages. You won't see any output — it's waiting for a client to connect. In the Testing section you'll see how to interact with it.

Anatomy of the server

mcp = FastMCP("demo-server")

Creates an instance with an identifying name. This name shows up when a client does discovery. Use descriptive names: "research-tools", "file-manager", "database-access".

@mcp.tool()
def add(a: int, b: int) -> int:
    """Add two integers."""
    return a + b

The @mcp.tool() decorator registers the function as an MCP tool. The SDK automatically extracts:

The same pattern as LangChain's @tool (module 2) — but now the tool lives in an independent server that any client can discover.

mcp.run(transport="stdio")

Launches the server using stdio: it reads from stdin, writes to stdout. The client launches your script as a subprocess and communicates through pipes. It's the simplest transport and the default.

Defining Tools

Tools with typed arguments

The type hints define the schema the client receives:

from mcp.server.fastmcp import FastMCP

mcp = FastMCP("research-tools")

@mcp.tool()
def search_papers(query: str, max_results: int = 5) -> str:
    """Search academic papers by topic.

    Returns titles and abstracts of relevant papers.
    The query should be specific: 'transformer attention mechanisms'
    instead of 'AI'.
    """
    papers = [
        {"title": "Attention Is All You Need", "year": 2017},
        {"title": "BERT: Pre-training of Deep Bidirectional Transformers", "year": 2018},
        {"title": "GPT-4 Technical Report", "year": 2023},
    ]
    results = papers[:max_results]
    output = f"Found {len(results)} papers for '{query}':\n"
    for i, p in enumerate(results, 1):
        output += f"  {i}. {p['title']} ({p['year']})\n"
    return output

@mcp.tool()
def calculate_citation_score(paper_id: str, include_self_citations: bool = False) -> str:
    """Compute a paper's citation score.

    paper_id: The paper's identifier (e.g.: 'arxiv:1706.03762')
    include_self_citations: If True, includes the author's self-citations.
    """
    base_score = 42500
    if include_self_citations:
        base_score += 3200
    return f"Paper {paper_id}: {base_score:,} citations"

The full docstring becomes the description. Parameters with defaults are optional in the schema.

Async tools with HTTP requests

The most useful tools call external APIs. Use async so you don't block the server:

import httpx
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("web-tools")

@mcp.tool()
async def fetch_webpage(url: str) -> str:
    """Download a web page's content. The url must include https://"""
    try:
        async with httpx.AsyncClient(timeout=15) as client:
            response = await client.get(url, follow_redirects=True)
            response.raise_for_status()
            content = response.text
            if len(content) > 5000:
                content = content[:5000] + "\n...[truncated]"
            return content
    except httpx.TimeoutException:
        return f"Error: timeout accessing {url}"
    except httpx.HTTPStatusError as e:
        return f"HTTP error {e.response.status_code} accessing {url}"
    except Exception as e:
        return f"Error: {type(e).__name__}: {e}"

Always include a timeout and error handling. A request with no timeout can hang the entire server.

Tools with validation: Pydantic Field

For richer schemas with per-field descriptions and constraints:

from pydantic import Field
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("database-tools")

@mcp.tool()
def query_database(
    table: str = Field(description="Table name: users, orders, products"),
    conditions: str = Field(description="WHERE conditions: 'status=active AND age>25'"),
    limit: int = Field(default=10, description="Maximum rows (1-100)", ge=1, le=100),
) -> str:
    """Run a read query. SELECT only, no INSERT/UPDATE/DELETE."""
    return f"SELECT * FROM {table} WHERE {conditions} LIMIT {limit}\n→ 3 rows"

A server with multiple tools from one domain

A good MCP server groups tools that share a domain. Don't create a server with a single tool (unnecessary overhead), or one with 50 unrelated tools.

from mcp.server.fastmcp import FastMCP
from datetime import datetime

mcp = FastMCP("notes-manager")
NOTES: dict[str, dict] = {}

@mcp.tool()
def create_note(title: str, content: str, tags: list[str] = []) -> str:
    """Create a new note. tags example: ['RAG', 'embeddings']"""
    note_id = f"note_{len(NOTES) + 1}"
    NOTES[note_id] = {
        "title": title, "content": content,
        "tags": tags, "created_at": datetime.now().isoformat(),
    }
    return f"Note created: {note_id} — '{title}'"

@mcp.tool()
def search_notes(query: str, tag_filter: str = "") -> str:
    """Search notes by content or title. An empty tag_filter = all."""
    results = []
    for nid, note in NOTES.items():
        matches_q = query.lower() in note["title"].lower() or query.lower() in note["content"].lower()
        matches_t = not tag_filter or tag_filter in note["tags"]
        if matches_q and matches_t:
            results.append(f"  [{nid}] {note['title']}")
    return "\n".join(results) if results else f"No results for '{query}'"

@mcp.tool()
def update_note(note_id: str, content: str) -> str:
    """Update a note. Use search_notes to find the ID."""
    if note_id not in NOTES:
        return f"Error: note '{note_id}' not found."
    NOTES[note_id]["content"] = content
    return f"Note '{note_id}' updated."

@mcp.tool()
def list_notes(limit: int = 20) -> str:
    """List every note sorted by date."""
    if not NOTES:
        return "No notes."
    lines = [f"  [{nid}] {n['title']} — {n['created_at'][:10]}" for nid, n in list(NOTES.items())[:limit]]
    return f"Total: {len(NOTES)}\n" + "\n".join(lines)

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

An agent that connects discovers 4 related tools and can use them in sequence: search_notes → update_note, or create_note → list_notes. Think in domains: "filesystem", "database", "research", "communications".

Defining Resources

Tools execute actions — resources expose data. A resource is a read endpoint identified by a URI. Think of it as a GET endpoint in REST: no side effects.

Basic @mcp.resource()

from mcp.server.fastmcp import FastMCP
from datetime import datetime

mcp = FastMCP("data-server")

@mcp.resource("config://app/settings")
def get_settings() -> str:
    """The application's current configuration."""
    return str({"model": "gpt-4o", "temperature": 0.7, "max_tokens": 4096})

@mcp.resource("status://server/health")
def get_health() -> str:
    """The server's health status."""
    return f"Status: healthy | Last check: {datetime.now().isoformat()}"

When a client requests config://app/settings, the server runs get_settings() and returns the result.

Resources with parameters (URI templates)

from mcp.server.fastmcp import FastMCP

mcp = FastMCP("paper-server")

PAPERS_DB = {
    "1706.03762": {"title": "Attention Is All You Need", "authors": "Vaswani et al.", "year": 2017},
    "1810.04805": {"title": "BERT", "authors": "Devlin et al.", "year": 2018},
}

@mcp.resource("papers://{paper_id}")
def get_paper(paper_id: str) -> str:
    """A paper's details by ID."""
    paper = PAPERS_DB.get(paper_id)
    if not paper:
        return f"Paper '{paper_id}' not found."
    return f"Title: {paper['title']}\nAuthors: {paper['authors']}\nYear: {paper['year']}"

@mcp.resource("papers://{paper_id}/citations")
def get_citations(paper_id: str) -> str:
    """A paper's citations."""
    citations = {"1706.03762": 95000, "1810.04805": 78000}
    return f"Paper {paper_id}: {citations.get(paper_id, 0):,} citations"

The {paper_id} gets extracted from the URI and passed as an argument.

Tools vs Resources: when to use each

Rule: if it modifies state → tool. If it only returns data → resource.

Defining Prompts

MCP prompts are reusable templates the server exposes. A client can list them and use them to build predefined conversations.

from mcp.server.fastmcp import FastMCP

mcp = FastMCP("research-prompts")

@mcp.prompt()
def analyze_paper(paper_title: str, paper_abstract: str) -> str:
    """Generate a prompt to analyze an academic paper."""
    return f"""Analyze the following academic paper:

Title: {paper_title}
Abstract: {paper_abstract}

Provide:
1. The main contribution
2. The methodology used
3. Key results
4. Limitations
5. Practical relevance"""

@mcp.prompt()
def compare_papers(paper_a: str, paper_b: str) -> str:
    """A prompt to compare two academic papers."""
    return f"""Compare these papers:

Paper A: {paper_a}
Paper B: {paper_b}

Criteria: technical approach, scale of the experiments, results,
practical applicability. Use a table format."""

@mcp.prompt()
def summarize_for_audience(content: str, audience: str = "general") -> str:
    """Summarize technical content for a specific audience."""
    instructions = {
        "general": "Avoid technical jargon. Use everyday analogies.",
        "technical": "Keep the technical terms. Include implementation details.",
        "executive": "Focus on business impact and ROI. 3 paragraphs maximum.",
    }
    inst = instructions.get(audience, instructions["general"])
    return f"Summarize for a {audience} audience:\n\n{content}\n\n{inst}"

When a client calls list_prompts(), it gets names, descriptions, and arguments. Then it invokes get_prompt("analyze_paper", ...) and receives the rendered template.

Configuring Transports

stdio: the default for local development

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

No network configuration. The client manages the process's lifecycle. It only works locally, one client per instance. Ideal for development and testing.

SSE: remote access

if __name__ == "__main__":
    mcp.run(transport="sse", host="0.0.0.0", port=8000)

Remote access, multiple simultaneous clients. Deployable as a service. It requires network configuration and manual authentication.

When to use each

Switching transports without touching the tools

import sys
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("flexible-server")

@mcp.tool()
def process_data(data: str) -> str:
    """Process text data."""
    return f"Processed: {data.upper()}"

if __name__ == "__main__":
    transport = sys.argv[1] if len(sys.argv) > 1 else "stdio"
    if transport == "sse":
        mcp.run(transport="sse", host="0.0.0.0", port=8000)
    else:
        mcp.run(transport="stdio")

python server.py          # stdio (default)
python server.py sse      # SSE on port 8000

Your tools' logic doesn't change — only the transport.

Testing the Server

mcp dev: the interactive inspector

mcp dev demo_server.py

Launches an inspector in the browser where you can:

1. See every registered tool with its schema
2. Invoke tools manually with arguments
3. List resources and read their content
4. See prompts and render them
5. Inspect the protocol's JSON-RPC messages

Before connecting a real client, always verify with mcp dev that your tools respond correctly.

An example for testing

from mcp.server.fastmcp import FastMCP
from datetime import datetime

mcp = FastMCP("test-server")

@mcp.tool()
def echo(message: str) -> str:
    """Return the message it received."""
    return f"Echo: {message}"

@mcp.tool()
def current_time() -> str:
    """The server's current date and time."""
    return datetime.now().isoformat()

@mcp.tool()
def word_count(text: str) -> str:
    """Count words, characters and lines."""
    return f"Words: {len(text.split())} | Characters: {len(text)}"

@mcp.resource("status://server")
def server_status() -> str:
    """The current status."""
    return f"test-server | running | {datetime.now().isoformat()}"

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

mcp dev test_server.py

In the inspector: go to Tools (3 tools), invoke echo with {"message": "hello MCP"}. Check Resources for status://server.

Connection to the Project

In this module's project (capsule 08), you'll build a Research Agent connected to 3 MCP servers:

1. Filesystem MCP Server — Tools to read/write research files + resources with file://{path}.
2. Web Search MCP Server — Replaces the hardcoded tool. You can update the server without touching the agent.
3. Paper Database MCP Server — Tools to search and manage papers. Similar to this capsule's notes-manager.

In the next capsule (04), you'll learn to build the client that connects your LangGraph agent to these servers for dynamic tool discovery.

The progression: M2 taught hardcoded tools → M3 function calling patterns → this capsule externalizes tools into MCP servers → capsule 04 consumes them from the agent.

Troubleshooting

Problem 1: "ModuleNotFoundError: No module named 'mcp'"

Cause: The SDK isn't installed, or the wrong virtual environment.

Solution: which python to check the environment, then pip install mcp.

Problem 2: "mcp dev doesn't show my tools"

Cause: Missing if __name__ == "__main__": mcp.run(), or there's a syntax error.

Solution: Verify that python server.py doesn't error. Make sure mcp.run(transport="stdio") is in the if __name__ block.

Problem 3: "My tool returns an empty response"

Cause: The function doesn't return a value, or it returns a non-serializable type.

Solution: Always return a str. If the result is a dict, use str(result).

Problem 4: "The server hangs with async tools"

Cause: An HTTP request with no timeout.

Solution: Always httpx.AsyncClient(timeout=10). Handle TimeoutException.

Problem 5: "Tools with empty schemas in the inspector"

Cause: Missing type hints on the parameters.

Solution: Every parameter needs a type hint: def tool(query: str, limit: int = 10) -> str:.

Exercises

Exercise 1: A server with 3 text tools (Easy)

Create "text-utils" with: to_uppercase(text), char_count(text), reverse(text). Test it with mcp dev.

from mcp.server.fastmcp import FastMCP

mcp = FastMCP("text-utils")

@mcp.tool()
def to_uppercase(text: str) -> str:
    """Convert text to uppercase."""
    return text.upper()

@mcp.tool()
def char_count(text: str) -> str:
    """Count the characters in a text."""
    return f"Total: {len(text)} ({len(text.replace(' ', ''))} without spaces)"

@mcp.tool()
def reverse(text: str) -> str:
    """Reverse the order of the characters."""
    return text[::-1]

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

Exercise 2: Add resources to the server (Easy)

To the previous server, add: stats://total-tools (returns the count) and help://{tool_name} (usage instructions).

TOOL_HELP = {
    "to_uppercase": "Send text → uppercase version. E.g.: 'hi' → 'HI'",
    "char_count": "Send text → counts characters with and without spaces.",
    "reverse": "Send text → reverses the characters. E.g.: 'abc' → 'cba'",
}

@mcp.resource("stats://total-tools")
def total_tools() -> str:
    """The number of available tools."""
    return f"Tools: 3 ({', '.join(TOOL_HELP.keys())})"

@mcp.resource("help://{tool_name}")
def tool_help(tool_name: str) -> str:
    """A tool's instructions."""
    return TOOL_HELP.get(tool_name, f"Tool '{tool_name}' not found.")

Exercise 3: A server with shared state (Medium)

Create "task-tracker" with: add_task(title, priority), complete_task(task_id), list_tasks(status) (status: "all"/"pending"/"completed"). A tasks://summary resource with the counts. Use an in-memory dict.

from mcp.server.fastmcp import FastMCP
from datetime import datetime

mcp = FastMCP("task-tracker")
TASKS: dict[str, dict] = {}

@mcp.tool()
def add_task(title: str, priority: str = "medium") -> str:
    """Add a task. priority: 'low', 'medium', 'high'."""
    tid = f"task_{len(TASKS) + 1}"
    TASKS[tid] = {"title": title, "priority": priority, "status": "pending",
                  "created_at": datetime.now().isoformat()}
    return f"Created: {tid} — '{title}' [{priority}]"

@mcp.tool()
def complete_task(task_id: str) -> str:
    """Mark a task as completed."""
    if task_id not in TASKS:
        return f"Error: '{task_id}' not found."
    TASKS[task_id]["status"] = "completed"
    return f"Completed: {TASKS[task_id]['title']}"

@mcp.tool()
def list_tasks(status: str = "all") -> str:
    """List tasks. status: 'all', 'pending', 'completed'."""
    if not TASKS:
        return "No tasks."
    filtered = {t: d for t, d in TASKS.items() if status == "all" or d["status"] == status}
    lines = [f"  {'✓' if d['status']=='completed' else '○'} [{t}] {d['title']}" for t, d in filtered.items()]
    return "\n".join(lines) if lines else f"No '{status}' tasks."

@mcp.resource("tasks://summary")
def summary() -> str:
    """A task summary."""
    p = sum(1 for t in TASKS.values() if t["status"] == "pending")
    return f"Total: {len(TASKS)} | Pending: {p} | Completed: {len(TASKS)-p}"

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

Exercise 4: Async tools with error handling (Medium)

Create "api-tools" with a fetch_github_repo(owner, repo) that queries the GitHub API and returns the name/stars/language. Handle timeouts, 404s, and generic errors with httpx.

import httpx
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("api-tools")

@mcp.tool()
async def fetch_github_repo(owner: str, repo: str) -> str:
    """Info about a GitHub repo. E.g.: owner='langchain-ai', repo='langchain'."""
    try:
        async with httpx.AsyncClient(timeout=10) as client:
            r = await client.get(f"https://api.github.com/repos/{owner}/{repo}")
            if r.status_code == 404:
                return f"'{owner}/{repo}' not found."
            r.raise_for_status()
            d = r.json()
            return f"{d['full_name']} | Stars: {d['stargazers_count']:,} | Lang: {d.get('language','N/A')}"
    except httpx.TimeoutException:
        return f"Timeout querying '{owner}/{repo}'."
    except Exception as e:
        return f"Error: {type(e).__name__}: {e}"

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

Exercise 5: A complete server with tools + resources + a prompt (Hard)

Create "research-assistant" with: 3 tools (search_papers, save_finding, list_findings), 2 resources (findings://{id}, findings://summary), and 1 prompt (research_plan(topic)). In-memory state. Test it with mcp dev.

from mcp.server.fastmcp import FastMCP
from datetime import datetime

mcp = FastMCP("research-assistant")
FINDINGS: dict[str, dict] = {}
PAPERS = [
    {"title": "Attention Is All You Need", "authors": "Vaswani et al.", "year": 2017},
    {"title": "BERT", "authors": "Devlin et al.", "year": 2018},
    {"title": "GPT-4 Technical Report", "authors": "OpenAI", "year": 2023},
]

@mcp.tool()
def search_papers(query: str, max_results: int = 3) -> str:
    """Search papers. A specific query: 'transformer attention' instead of 'AI'."""
    matches = [p for p in PAPERS if query.lower() in p["title"].lower()] or PAPERS
    lines = [f"  {i}. {p['title']} — {p['authors']} ({p['year']})" for i, p in enumerate(matches[:max_results], 1)]
    return f"Results ({len(lines)}):\n" + "\n".join(lines)

@mcp.tool()
def save_finding(title: str, content: str, source: str = "manual") -> str:
    """Save a research finding."""
    fid = f"finding_{len(FINDINGS) + 1}"
    FINDINGS[fid] = {"title": title, "content": content, "source": source,
                     "saved_at": datetime.now().isoformat()}
    return f"Saved: {fid} — '{title}'"

@mcp.tool()
def list_findings() -> str:
    """List the saved findings."""
    if not FINDINGS:
        return "No findings. Use save_finding."
    lines = [f"  [{f}] {d['title']} — {d['source']}" for f, d in FINDINGS.items()]
    return "\n".join(lines)

@mcp.resource("findings://{finding_id}")
def get_finding(finding_id: str) -> str:
    """A finding's details."""
    f = FINDINGS.get(finding_id)
    if not f:
        return f"'{finding_id}' not found."
    return f"Title: {f['title']}\nSource: {f['source']}\n\n{f['content']}"

@mcp.resource("findings://summary")
def findings_summary() -> str:
    """A summary of the findings."""
    if not FINDINGS:
        return "No findings."
    sources = {}
    for f in FINDINGS.values():
        sources[f["source"]] = sources.get(f["source"], 0) + 1
    return f"Total: {len(FINDINGS)} | {', '.join(f'{s}:{c}' for s,c in sources.items())}"

@mcp.prompt()
def research_plan(topic: str) -> str:
    """A structured research plan."""
    return f"""Research plan: {topic}

1. Goal: What to discover about {topic}
2. Key questions: 3-5 specific questions
3. Methodology: search_papers → save_finding → synthesize
4. Deliverable: The result's format

Start with search_papers on '{topic}'."""

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

Summary

In this capsule you built your first working MCP server:

• FastMCP is the high-level interface: FastMCP("name") + decorators
• @mcp.tool() turns functions into MCP tools. The name, docstring, and type hints become a JSON Schema
• @mcp.resource() exposes read data via URIs with parameters ({variable}). Tools for actions, resources for data
• @mcp.prompt() registers reusable templates the client can list and render
• Transports: stdio for local development, SSE for remote access. Your tool code doesn't change
• mcp dev is the visual inspector for verifying everything before connecting a client
• Async tools for I/O — always with timeouts
• Group tools by coherent domain: one server per function

Next capsule: MCP Clients — you'll connect an agent to your server, do dynamic tool discovery, and integrate them as LangChain tools in your StateGraph.

Additional Resources

1. MCP Python SDK — GitHub — The SDK's official repository
2. MCP Servers Documentation — The official server guide
3. MCP Spec — Tools — The formal tools specification
4. MCP Spec — Resources — The formal resources specification
5. MCP Inspector — The inspector's documentation
6. httpx Documentation — The async HTTP client for network tools