D2. Building an MCP Server: Wrapping Custom Tools for Any MCP Client

About this article This article is part of Building with Claude — A Practitioner's Guide to the Anthropic API, a study-notes-plus-commentary series based on Anthropic's official "Building with the Claude API" course (hosted on Coursera) and the public Anthropic API documentation at docs.anthropic.com.

Original course and documentation material is © Anthropic. Direct quotes are cited inline. MCP specification material is from modelcontextprotocol.io. Commentary, code adaptations, and examples are © DataMy. This series is independent and not affiliated with or endorsed by Anthropic.

Companion notebook: D2_mcp_server.ipynb Setup: see README.md — requires mcp>=1.0.0 Server source: mcp_data_server.py (walkthrough + extension in the notebook) CLI app: scripts/data_monitor_cli.py

---

The server perspective

D1 showed MCP from the client side: connecting to a running server, discovering its capabilities, and using Claude to interact with them. D2 flips the perspective.

Building an MCP server is the act of packaging domain knowledge — what tools exist, how they are described, what data they expose — into a reusable interface. Any MCP client that connects to your server gets access to that knowledge without reimplementing it. The more precisely you design the server's interface, the better Claude (or any other client) performs when using it.

This article walks through mcp_data_server.py in detail, explains the design decisions behind each tool, resource, and prompt, extends the server with a new tool, and then builds a standalone CLI application that uses the server.

---

1. FastMCP: the server framework

The mcp Python library provides two APIs for building servers:

Low-level API (mcp.server.Server): explicit handler registration, maximum control, more boilerplate. Appropriate when you need custom protocol-level behaviour.

FastMCP (mcp.server.fastmcp.FastMCP): decorator-based, high-level, production-ready for most use cases. FastMCP reads Python type hints and docstrings to generate the MCP tool schemas automatically. This is what mcp_data_server.py uses.

The minimum viable FastMCP server:

from mcp.server.fastmcp import FastMCP

mcp = FastMCP("my-server")

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

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

FastMCP does three things automatically:

1. Generates the JSON Schema inputSchema from the function's type hints
2. Uses the function's docstring as the tool description
3. Wraps the return value in the MCP CallToolResult response structure

This means your tool function is just a Python function with a good docstring. No boilerplate protocol code.

---

2. Tool design from the server author's view

In C2, tool design was about writing schemas that help Claude select the right tool. As the server author, the same principles apply — but your audience is broader. Claude is one consumer; IDE extensions, other AI systems, and your own CLI applications are others. The description is the contract with all of them.

The three tools in mcp_data_server.py each address a different monitoring question:

get_warehouse_summary(days) — "What is the current cost distribution?" This is a read-across question: it needs all warehouses for a time window. The description names the triggering intents explicitly ("understand current cost distribution", "identify the most expensive warehouses") and tells clients when to call alongside other tools ("For cost trend analysis and anomaly detection").

get_failing_jobs(days) — "What is currently broken?" The description distinguishes this from a warehouse cost query by naming the triggering intents ("pipeline health", "failure clusters") and gives a cross-tool hint: "For correlating failures with warehouse cost spikes, call alongside get_warehouse_summary."

check_job_sla(job_name) — "Is this specific table up to date?" The description gives the concrete triggering phrases ("is this table up to date?", "when did X last run successfully?") and names the SLA definition (26 hours) so clients can interpret the result without needing to know the business rule.

Schema generation from type hints

FastMCP converts Python type hints to JSON Schema:

@mcp.tool()
def get_warehouse_summary(days: int = 7) -> str:
    ...

Produces the input schema:

{
  "type": "object",
  "properties": {
    "days": {
      "type": "integer",
      "default": 7
    }
  }
}

The default value in the Python signature becomes the JSON Schema default. Optional parameters (with defaults) are not in required. Required parameters (no default) are added to required automatically. This means you do not need to write JSON Schema manually for most tools.

For parameters that need additional constraints (enum values, descriptions, min/max), use Annotated from the standard library:

from typing import Annotated
from pydantic import Field

@mcp.tool()
def get_warehouse_summary(
    days: Annotated[int, Field(ge=1, le=90, description="Days to look back")] = 7
) -> str:
    ...

---

3. Resources: exposing data, not functions

Resources are the right abstraction for content that is:

• Reference material (runbooks, schema definitions, configuration)
• Relatively static — updated weekly or less frequently, not per-request
• Consumed as whole documents rather than computed summaries

A function that returns static text as a tool call is a resource waiting to be extracted. The practical difference: tools appear in the tool use loop (Claude actively calls them, billing output tokens for each decision); resources are fetched explicitly by the client and injected as context.

@mcp.resource("data://warehouse-runbook")
def get_warehouse_runbook() -> str:
    """The Snowflake warehouse cost runbook."""
    return Path("../data/runbook_warehouse_cost.md").read_text()

The URI scheme (data://) is arbitrary — choose one that makes sense for your domain. Common conventions:

• file:// for file system resources
• db://table-name for database resources
• config://setting-name for configuration resources
• Custom schemes (data://, docs://, api://) for application-specific resources

When to use a resource vs a tool

Characteristic	Use a resource	Use a tool
Content changes	Weekly or less	Daily or more often
Computation required	No (read static file)	Yes (query, aggregate, filter)
Size	Document-scale (KB to low MB)	Response-scale (bytes to KB)
Client usage	Inject as context	Execute in tool use loop
Analogy	RAG document store	API endpoint

The runbook in mcp_data_server.py is a resource: it is a markdown file that changes when the data team updates procedures, not when someone asks a question. The warehouse summary is a tool: it computes a different result for each days argument.

---

4. Prompts: server-side query patterns

Prompts centralise how clients should ask Claude about your domain. Think of them as the "recommended questions" that the server author has tested and knows produce good results.

@mcp.prompt()
def incident_analysis(warehouse_name: str, date_range: str) -> str:
    """Structured prompt for a Snowflake cost incident analysis."""
    return f"""You are a Snowflake cost analyst...

Warehouse  : {warehouse_name}
Date range : {date_range}

Perform the following analysis steps in order:
1. Call get_warehouse_summary...
2. Identify days where spend exceeded 1.25x median...
...
"""

FastMCP reads the function's parameters as the prompt arguments, and its docstring as the prompt description. Clients retrieve the prompt by name with session.get_prompt("incident_analysis", arguments={...}) and receive the rendered string.

Prompts are especially useful for multi-step agentic workflows where the order of tool calls matters. The incident_analysis prompt instructs Claude to call tools in a specific sequence (cost data, then failure data, then runbook lookup, then SLA check) that produces a more coherent incident report than an unconstrained question would.

---

5. Adding a new tool: get_query_performance

One of the key benefits of MCP is that server improvements are immediately available to all clients. Adding a tool requires only a change to the server file — no client code changes.

The D2 notebook extends mcp_data_server.py with a fourth tool that surfaces slow-query patterns from the warehouse usage data:

@mcp.tool()
def get_query_performance(warehouse_name: str, days: int = 14) -> str:
    """Return average execution and queue time trends for a named warehouse.

    Call this when investigating whether a warehouse is experiencing query slowdowns,
    queue buildup, or execution time regression. Use alongside get_warehouse_summary
    to correlate performance degradation with credit cost increases.

    Args:
        warehouse_name: Exact warehouse name (e.g. WH_BI_M).
        days: Number of days to look back.
    """
    rows = [
        r for r in USAGE_ROWS
        if r["warehouse_name"] == warehouse_name
    ]
    if not rows:
        all_wh = sorted({r["warehouse_name"] for r in USAGE_ROWS})
        return f"Warehouse '{warehouse_name}' not found. Available: {', '.join(all_wh)}"

    all_dates = sorted({r["date"] for r in USAGE_ROWS})
    # days-1 so "last N days" returns exactly N rows (timedelta(days=N) with >= gives N+1)
    cutoff = (
        datetime.strptime(all_dates[-1], "%Y-%m-%d") - timedelta(days=days - 1)
    ).strftime("%Y-%m-%d")
    recent = [r for r in rows if r["date"] >= cutoff]

    if not recent:
        return f"No data for {warehouse_name} in the last {days} days (since {cutoff})."

    lines = [
        f"Query performance -- {warehouse_name} -- last {days} days",
        "",
        f"{'Date':<12} {'Queries':>8} {'Avg Queue (s)':>14} {'Avg Exec (s)':>13}",
        "-" * 52,
    ]
    for r in sorted(recent, key=lambda x: x["date"]):
        lines.append(
            f"{r['date']:<12} {int(r['query_count']):>8} "
            f"{float(r['avg_queue_time_s']):>14.2f} "
            f"{float(r['avg_execution_time_s']):>13.2f}"
        )
    avg_queue = sum(float(r["avg_queue_time_s"]) for r in recent) / len(recent)
    avg_exec  = sum(float(r["avg_execution_time_s"]) for r in recent) / len(recent)
    lines.append("-" * 52)
    lines.append(f"{'Period average':<12} {'':>8} {avg_queue:>14.2f} {avg_exec:>13.2f}")
    return "\n".join(lines)

After adding this function to mcp_data_server.py, the next client connection will automatically discover four tools via session.list_tools(). No changes to the D1 notebook or any other client are required.

---

6. Building a CLI application

The practical output of D1 + D2 is a reusable MCP server that any application can connect to. A CLI application is the most concrete example: a terminal tool that accepts a question and returns Claude's answer, backed by the MCP server's tools.

The CLI pattern:

#!/usr/bin/env python3
"""data_monitor_cli.py -- CLI monitoring agent using the MCP data platform server.

Usage: python data_monitor_cli.py "Which warehouse has the highest cost this week?"
       python data_monitor_cli.py --prompt incident_analysis --warehouse WH_BI_M
"""
import argparse
import asyncio
import sys
from pathlib import Path

import anthropic
import nest_asyncio
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

nest_asyncio.apply()

MODEL = "claude-sonnet-4-5"
SERVER_SCRIPT = Path(__file__).resolve().parent.parent / "notebooks" / "mcp_data_server.py"
SERVER_PARAMS = StdioServerParameters(command=sys.executable, args=[str(SERVER_SCRIPT)])


async def run_agent(question: str) -> str:
    anthropic_client = anthropic.Anthropic()
    async with stdio_client(SERVER_PARAMS) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()
            tools_result = await session.list_tools()
            anthropic_tools = [
                {"name": t.name, "description": t.description or "",
                 "input_schema": t.inputSchema}
                for t in tools_result.tools
            ]
            messages = [{"role": "user", "content": question}]
            for _ in range(10):
                response = anthropic_client.messages.create(
                    model=MODEL, max_tokens=4096,
                    tools=anthropic_tools, messages=messages,
                )
                if response.stop_reason == "end_turn":
                    return next(
                        (b.text for b in response.content if b.type == "text"),
                        "(no text response)",
                    )
                if response.stop_reason == "tool_use":
                    messages.append({"role": "assistant", "content": response.content})
                    results = []
                    for block in response.content:
                        if block.type == "tool_use":
                            r = await session.call_tool(block.name, block.input)
                            results.append({
                                "type": "tool_result",
                                "tool_use_id": block.id,
                                "content": r.content[0].text,
                            })
                    messages.append({"role": "user", "content": results})
    return "(max turns reached)"


def main() -> None:
    parser = argparse.ArgumentParser(description="Data platform monitoring CLI")
    parser.add_argument("question", nargs="?", help="Question to ask")
    parser.add_argument("--prompt", help="Use a server-side prompt template by name")
    parser.add_argument("--warehouse", default="WH_BI_M",
                        help="Warehouse name (for prompt templates)")
    parser.add_argument("--date-range", default="2025-07-01 to 2025-09-28",
                        help="Date range (for prompt templates)")
    args = parser.parse_args()

    if args.prompt:
        async def get_prompt_text():
            async with stdio_client(SERVER_PARAMS) as (read, write):
                async with ClientSession(read, write) as session:
                    await session.initialize()
                    result = await session.get_prompt(
                        args.prompt,
                        arguments={
                            "warehouse_name": args.warehouse,
                            "date_range": args.date_range,
                        },
                    )
                    return result.messages[0].content.text
        question = asyncio.get_event_loop().run_until_complete(get_prompt_text())
    elif args.question:
        question = args.question
    else:
        parser.print_help()
        sys.exit(1)

    print(f"Agent running...\n{'=' * 60}", flush=True)
    answer = asyncio.get_event_loop().run_until_complete(run_agent(question))
    print(answer)


if __name__ == "__main__":
    main()

Usage examples:

# Direct question
python data_monitor_cli.py "Which job has the most SLA breaches this month?"

# Server-side prompt template
python data_monitor_cli.py --prompt incident_analysis --warehouse WH_BI_M \
    --date-range "2025-07-20 to 2025-08-10"

The CLI is thin by design: it handles argument parsing and async plumbing, while the domain knowledge lives in the MCP server. A new question is just a different argument; a new tool is a new decorator in mcp_data_server.py. The application layer does not need to change.

---

7. Transport: stdio vs HTTP/SSE

All the code in D1 and D2 uses stdio transport: the client launches the server as a subprocess and communicates over stdin/stdout. This is the right choice for:

• Local development and testing
• Single-client, single-server deployments (one CLI application, one notebook)
• Serverless use cases where the server starts fresh per invocation

For production deployments with multiple clients, a persistent server, or remote access, HTTP/SSE transport is appropriate. FastMCP supports it with a single flag:

# Stdio (default):  mcp.run()
# HTTP/SSE:
mcp.run(transport="sse", host="0.0.0.0", port=8000)

The client side switches from stdio_client to an SSE client:

from mcp.client.sse import sse_client
async with sse_client("http://my-server:8000/sse") as (read, write):
    ...

Everything else — ClientSession, initialize(), list_tools(), call_tool() — is identical. The transport is an implementation detail hidden behind the same session interface.

"MCP defines two standard transport mechanisms: Standard Input/Output (stdio) for local integrations and HTTP with Server-Sent Events (SSE) for remote integrations." — MCP specification, "Core architecture — Transports", modelcontextprotocol.io, accessed 2026-06-10.

---

8. Server design checklist

Before shipping an MCP server to other teams:

Tools

• Every tool description answers: when to call it, what it does, what inputs it expects
• Cross-tool hints are in the descriptions ("call alongside X for Y")
• Parameters have defaults where sensible; required parameters have clear descriptions
• Tool output is self-contextualised (includes baselines, labels, anomaly flags)
• Error cases return informative strings, not Python exceptions

Resources

• URIs follow a consistent scheme (data://, config://, etc.)
• Resource content is self-contained (clients should not need to call a tool to interpret a resource)
• Large resources (>500KB) are candidates for chunking or pagination

Prompts

• Prompt arguments have clear names and descriptions
• The prompt text has been tested with the tools it references
• The prompt produces a coherent output when given to Claude without additional context

Operations

• The server handles missing datasets gracefully (informative error, not a crash)
• The server starts in under 2 seconds (defer heavy work to tool calls)
• The server is tested with python server.py before being connected to a client

---

Practitioner Notes

• Keep the server stateless. Each tool call should produce the same result for the same inputs, regardless of previous calls. Stateful servers are harder to test, harder to restart, and harder to reason about when debugging agent behaviour.
• Load datasets at startup, not per call. The USAGE_ROWS and JOBS_ROWS module-level variables in mcp_data_server.py load the CSVs once when the server starts. Loading on every tool call adds latency and I/O overhead. For databases, use a connection pool.
• Test tools independently before testing with Claude. Run the server directly (python mcp_data_server.py) and call list_tools() from a simple async script. Verify each tool's output with a known input before attaching Claude. Bugs in tool output produce silent reasoning errors.
• Use server.run(transport="sse") for shared team servers. If multiple people or applications will use the same server, the HTTP/SSE transport avoids the per-client subprocess overhead and enables centralised logging.
• Version your server. FastMCP passes the version parameter to clients during initialization. Use semantic versioning ("1.0.0", "1.1.0") so clients can detect breaking changes. Pass it to FastMCP("server-name", version="1.0.0").

---

Beyond the Docs

The series ends at D2, but the pattern does not. An MCP server is a reusable AI interface layer on top of any data system — a Snowflake account, a Postgres database, a REST API, a file share. Once built and tested, it can serve:

• Claude in the API (D1 pattern)
• Claude Desktop (via the desktop MCP client configuration)
• IDE extensions (Cursor, Zed, VS Code with an MCP plugin)
• Automated monitoring CLI tools (D2 pattern)
• Multi-agent workflows where one Claude orchestrates another via MCP

The Anthropic documentation describes this as the direction of agentic AI:

"In agentic contexts, Claude will sometimes act as an orchestrator of multi-agent pipelines and sometimes as a subagent within those pipelines... MCP provides a standardized way to connect Claude to the tools and data sources it needs in those pipelines." — Anthropic API Docs, "Build with Claude — Agentic and multi-agent frameworks", accessed 2026-06-10.

Building an MCP server for your domain is not a Claude-specific investment. It is infrastructure for any AI system that follows the protocol. The time spent designing good tool descriptions, clean resource URIs, and tested prompt templates pays dividends every time a new client connects.

---

Previous: D1 — MCP Concepts & Using an MCP Server as a Client

Series complete. Return to series index: Building with the Claude API — A Practitioner's Guide

---

Based on Anthropic's "Building with the Claude API" course (Coursera), public API documentation at docs.anthropic.com, and the Model Context Protocol specification at modelcontextprotocol.io. Commentary © 2026 DataMy. Not affiliated with Anthropic.