Implement an MCP server with typed tool contracts

An MCP tool is a contract between your code and every model that calls it. The model only sees the tool’s name, description and JSON schemas, so those must say exactly what the tool accepts and returns. Here you implement a server for a software project’s task tracker with the official MCP Python SDK. Its three tools, list_tasks, add_task, complete_task, take constrained arguments and return Pydantic models. The SDK turns both into published schemas, checks every call against the input schema, and reports failures in two different ways: as tool errors the model can read, or as protocol errors the client raises.

When you need this. A release assistant files and closes tickets in your team’s task tracker while you talk to it. The model fills in the arguments for add_task and complete_task from the conversation, so a vague request can turn into a malformed call. A typed contract lets the server tell the model exactly what it accepts, and refuse anything else before your code runs.

The drawing follows request checks, argument validation, tool execution and output conversion. The SDK first checks that the request fits the protocol: a request with no tool name fails the protocol check as a JSON-RPC error, before tool code runs. Then it validates the arguments against the tool’s input model. Only valid arguments reach your function, and its return value is converted to structuredContent. A failure at the argument check or inside the function becomes a result with isError: true, so the model can read it and try again.

1. Set up the project

The server needs one package, the MCP Python SDK, which brings Pydantic with it. Download the complete lab ZIP, which includes the Python files, fixtures, tests and README. Extract it before running the commands below.

unzip implement-an-mcp-server-with-typed-tool-contracts.zip
cd mcp-typed-tools
python3.12 -m venv .venv
.venv/bin/python -m pip install -r requirements-lock.txt

2. Write the typed tools

Now we write the tools themselves. MCPServer is the SDK’s high-level server. Each function decorated with @server.tool() becomes a tool: the docstring becomes its description, the parameter annotations become its inputSchema, and the return annotation becomes its outputSchema.

"""An MCP server for a project task tracker, with typed tool contracts. Run over stdio: python tasks_server.py"""
from typing import Annotated, Literal

from pydantic import BaseModel, Field

from mcp.server import MCPServer
from mcp.server.mcpserver.exceptions import ToolError
from mcp.types import ToolAnnotations

Status = Literal["open", "done"]
Priority = Literal["low", "normal", "high"]
TaskId = Annotated[str, Field(pattern=r"^T-\d+$", description="A task ID such as T-1, as returned by list_tasks.")]


class Task(BaseModel):
    """One task in the tracker."""
    id: str = Field(description="Task ID such as T-1.")
    title: str
    priority: Priority
    status: Status


class TaskList(BaseModel):
    """The tasks that match a filter."""
    tasks: list[Task]
    count: int


server = MCPServer("tasks", version="1.0.0", instructions="Read and update the task tracker of one software project.")

TASKS: dict[str, Task] = {task.id: task for task in [
    Task(id="T-1", title="Write the release notes", priority="normal", status="open"),
    Task(id="T-2", title="Fix the login redirect bug", priority="high", status="done"),
    Task(id="T-3", title="Update the API docs", priority="low", status="open"),
]}


@server.tool(annotations=ToolAnnotations(read_only_hint=True))
def list_tasks(status: Status | None = None) -> TaskList:
    """List the project's tasks. Pass status to return only open or only done tasks."""
    tasks = [task for task in TASKS.values() if status is None or task.status == status]
    return TaskList(tasks=tasks, count=len(tasks))


@server.tool()
def add_task(title: Annotated[str, Field(min_length=3, max_length=80)], priority: Priority = "normal") -> Task:
    """Add an open task and return it with its new ID."""
    task = Task(id=f"T-{len(TASKS) + 1}", title=title, priority=priority, status="open")
    TASKS[task.id] = task
    return task


@server.tool(annotations=ToolAnnotations(idempotent_hint=True))
def complete_task(task_id: TaskId) -> Task:
    """Mark one task as done and return it."""
    if task_id not in TASKS:
        raise ToolError(f"No task with ID {task_id}. Call list_tasks to see the valid IDs.")
    TASKS[task_id].status = "done"
    return TASKS[task_id]


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

Each part of the signature is a decision about the contract:

• Literal types are enums. Status and Priority become JSON Schema enum lists, so the model sees the allowed values and the SDK refuses any other value.
• Annotated[str, Field(...)] adds constraints. min_length and max_length bound a title, and the pattern on TaskId accepts only IDs such as T-1. The description on the field tells the model where valid IDs come from.
• Return a model, not a dict. A Task return type gives the tool an outputSchema, and every result carries structuredContent that matches it, plus the same JSON as text for clients that read only text. list_tasks returns a TaskList object; the SDK would wrap a bare list[Task] in a {"result": [...]} object.
• ToolError is for failures the caller can fix. Its message goes back to the model in a normal result marked isError: true.
• Annotations are hints. read_only_hint and idempotent_hint tell a host that list_tasks changes nothing and that repeating complete_task has no further effect. The specification says clients must treat them as untrusted unless the server is trusted, so they do not replace a permission check.

The task list lives in memory so that every run starts from the same three tasks. A real server would read and write a database or an issue tracker’s API.

Arguments outside the input schema. The SDK returns isError: true with the Pydantic message, such as String should have at least 3 characters, before the function changes the task list. Correct the arguments using the published schema.

3. Read the published contract

Before calling anything, look at what the model will see. The schemas are what a host gives the model, so read them as the model does. record.py saves the tools/list result as it arrives over the wire. This is the complete_task definition:

{
  "name": "complete_task",
  "description": "Mark one task as done and return it.",
  "inputSchema": {
    "type": "object",
    "properties": {
      "task_id": {
        "description": "A task ID such as T-1, as returned by list_tasks.",
        "pattern": "^T-\\d+$",
        "title": "Task Id",
        "type": "string"
      }
    },
    "required": [
      "task_id"
    ],
    "title": "complete_taskArguments"
  },
  "outputSchema": {
    "description": "One task in the tracker.",
    "properties": {
      "id": {
        "description": "Task ID such as T-1.",
        "title": "Id",
        "type": "string"
      },
      "title": {
        "title": "Title",
        "type": "string"
      },
      "priority": {
        "enum": [
          "low",
          "normal",
          "high"
        ],
        "title": "Priority",
        "type": "string"
      },
      "status": {
        "enum": [
          "open",
          "done"
        ],
        "title": "Status",
        "type": "string"
      }
    },
    "required": [
      "id",
      "title",
      "priority",
      "status"
    ],
    "title": "Task",
    "type": "object"
  },
  "annotations": {
    "idempotentHint": true
  }
}

The pattern, the required list and the full Task shape all come from type hints. When you change a type, the contract changes with it, which is why the next lesson tests the published schemas directly.

4. Call the tools from a client

Client accepts several targets: a command to start over stdio, a URL for Streamable HTTP, or, as here, the MCPServer object itself. The in-process connection carries the same requests and results as a transport, without a subprocess. That keeps this lesson about contracts; the next lesson serves the same server over both transports.

"""Connect to the tasks server in-process, list its tools and make four calls. Run: python client.py"""
import asyncio

from mcp import Client, types
from mcp.shared.exceptions import MCPError

from tasks_server import server


async def main() -> None:
    async with Client(server) as client:
        tools = await client.list_tools()
        for tool in tools.tools:
            print(f"tool   {tool.name}({', '.join(tool.input_schema['properties'])}) -> {tool.output_schema['title']}")

        result = await client.call_tool("add_task", {"title": "Tag v1.2", "priority": "high"})
        print("ok    ", result.structured_content)

        result = await client.call_tool("add_task", {"title": "v2"})
        print("error ", result.is_error, result.content[0].text)

        result = await client.call_tool("complete_task", {"task_id": "T-9"})
        print("error ", result.is_error, result.content[0].text)

        try:  # a tools/call request with no tool name breaks the protocol, not the tool
            await client.session.send_request(types.Request(method="tools/call", params={"arguments": {}}), types.CallToolResult)
        except MCPError as error:
            print("raised", type(error).__name__, error.code, error.message)


if __name__ == "__main__":
    asyncio.run(main())

.venv/bin/python client.py

tool   list_tasks(status) -> TaskList
tool   add_task(title, priority) -> Task
tool   complete_task(task_id) -> Task
ok     {'id': 'T-4', 'title': 'Tag v1.2', 'priority': 'high', 'status': 'open'}
error  True Error executing tool add_task: 1 validation error for add_taskArguments
title
  String should have at least 3 characters [type=string_too_short, input_value='v2', input_type=str]
    For further information visit https://errors.pydantic.dev/2.14/v/string_too_short
error  True Error executing tool complete_task: No task with ID T-9. Call list_tasks to see the valid IDs.
raised MCPError -32602 Invalid request parameters

The recorded output above shows this exchange. The good call returned {"id":"T-4","title":"Tag v1.2","priority":"high","status":"open"} as structuredContent. The two bad calls came back as results with isError true, and the client did not raise. The last request had no tool name, so the client raised MCPError with code -32602.

Unknown task ID. ToolError becomes an isError: true result; the connection stays usable. The recorded message, Error executing tool complete_task: No task with ID T-9. Call list_tasks to see the valid IDs., names the tool that supplies valid IDs, so the model can look them up before retrying.

Unknown tool name. mcp 2.3.0 returns isError: true with Unknown tool: delete_task. The specification instead describes this as a protocol error, so a client talking to other SDKs must also catch MCPError.

Malformed request. A request without a tool name raises MCPError with code -32602 instead of returning a tool result. Fix the request envelope in the client; changing the tool’s arguments cannot repair it.

5. Test the contract

With the contract visible, we can pin it. The tests reload the server module before each test, so every test starts with the same three tasks. They check the published schemas, validate structuredContent against each outputSchema with jsonschema, and pin each refusal described in the steps above.

.venv/bin/python -m unittest -v test_tasks_server.py

test_client_script_prints_the_recorded_output (test_tasks_server.TypedTools.test_client_script_prints_the_recorded_output) ... ok
test_invalid_arguments_are_rejected_before_the_tool_runs (test_tasks_server.TypedTools.test_invalid_arguments_are_rejected_before_the_tool_runs) ... Tool 'add_task' rejected arguments: ['title']
Tool 'add_task' rejected arguments: ['priority']
Tool 'complete_task' rejected arguments: ['task_id']
ok
test_malformed_request_is_a_protocol_error (test_tasks_server.TypedTools.test_malformed_request_is_a_protocol_error) ... ok
test_saved_calls_match_a_fresh_run (test_tasks_server.TypedTools.test_saved_calls_match_a_fresh_run) ... Tool 'add_task' rejected arguments: ['title']
Tool 'complete_task' failed: 'Error executing tool complete_task: No task with ID T-9. Call list_tasks to see the valid IDs.'
ok
test_structured_content_matches_the_output_schema (test_tasks_server.TypedTools.test_structured_content_matches_the_output_schema) ... ok
test_tools_publish_input_and_output_schemas (test_tasks_server.TypedTools.test_tools_publish_input_and_output_schemas) ... ok
test_unexpected_exception_hides_its_details (test_tasks_server.TypedTools.test_unexpected_exception_hides_its_details) ... ok
test_unknown_task_id_is_a_tool_error_and_the_server_keeps_running (test_tasks_server.TypedTools.test_unknown_task_id_is_a_tool_error_and_the_server_keeps_running) ... Tool 'complete_task' failed: 'Error executing tool complete_task: No task with ID T-9. Call list_tasks to see the valid IDs.'
ok
test_unknown_tool_name_is_a_tool_error_in_this_sdk (test_tasks_server.TypedTools.test_unknown_tool_name_is_a_tool_error_in_this_sdk) ... Tool 'delete_task' failed: 'Unknown tool: delete_task'
ok

----------------------------------------------------------------------
Ran 9 tests in 0.423s

OK

All 9 tests pass. The lines that start with Tool are the server’s own log messages for the expected failures; the SDK logs the names of rejected fields, not their values.

"""Contract and failure tests for tasks_server.py. Run: python -m unittest -v test_tasks_server.py"""
import asyncio
import importlib
import json
import subprocess
import sys
import unittest
from pathlib import Path

from jsonschema import validate
from mcp import Client, types
from mcp.shared.exceptions import MCPError

import record
import tasks_server

HERE = Path(__file__).resolve().parent


class TypedTools(unittest.TestCase):
    def setUp(self):
        self.module = importlib.reload(tasks_server)  # a fresh in-memory task list for every test

    def run_client(self, work):
        async def main():
            async with Client(self.module.server) as client:
                return await work(client)
        return asyncio.run(main())

    def call(self, name, arguments):
        return self.run_client(lambda client: client.call_tool(name, arguments))

    def test_tools_publish_input_and_output_schemas(self):
        tools = {tool.name: tool for tool in self.run_client(lambda client: client.list_tools()).tools}
        self.assertEqual(list(tools), ["list_tasks", "add_task", "complete_task"])
        self.assertEqual(tools["add_task"].input_schema["properties"]["title"]["minLength"], 3)
        self.assertEqual(tools["complete_task"].input_schema["properties"]["task_id"]["pattern"], r"^T-\d+$")
        self.assertEqual(tools["list_tasks"].output_schema["title"], "TaskList")

    def test_structured_content_matches_the_output_schema(self):
        tools = {tool.name: tool for tool in self.run_client(lambda client: client.list_tools()).tools}
        for name, arguments in [("list_tasks", {"status": "open"}), ("add_task", {"title": "Tag v1.2"})]:
            result = self.call(name, arguments)
            self.assertFalse(result.is_error)
            validate(result.structured_content, tools[name].output_schema)
            self.assertEqual(json.loads(result.content[0].text), result.structured_content)

    def test_invalid_arguments_are_rejected_before_the_tool_runs(self):
        for name, arguments, message in [
            ("add_task", {"title": "v2"}, "String should have at least 3 characters"),
            ("add_task", {"title": "Tag v1.2", "priority": "urgent"}, "Input should be 'low', 'normal' or 'high'"),
            ("complete_task", {"task_id": "3"}, "String should match pattern"),
        ]:
            result = self.call(name, arguments)
            self.assertTrue(result.is_error)
            self.assertIn(message, result.content[0].text)
        self.assertEqual(len(self.module.TASKS), 3)

    def test_unknown_task_id_is_a_tool_error_and_the_server_keeps_running(self):
        async def work(client):
            return await client.call_tool("complete_task", {"task_id": "T-9"}), await client.call_tool("list_tasks", {})
        failed, listed = self.run_client(work)
        self.assertTrue(failed.is_error)
        self.assertIn("No task with ID T-9. Call list_tasks", failed.content[0].text)
        self.assertEqual(listed.structured_content["count"], 3)

    def test_unknown_tool_name_is_a_tool_error_in_this_sdk(self):
        result = self.call("delete_task", {"task_id": "T-1"})
        self.assertTrue(result.is_error)
        self.assertEqual(result.content[0].text, "Unknown tool: delete_task")

    def test_unexpected_exception_hides_its_details(self):
        self.module.TASKS["T-2"] = None  # a corrupt record makes list_tasks raise AttributeError
        with self.assertLogs("mcp.server.mcpserver.server", "ERROR"):
            result = self.call("list_tasks", {})
        self.assertTrue(result.is_error)
        self.assertEqual(result.content[0].text, "Error executing tool list_tasks")

    def test_malformed_request_is_a_protocol_error(self):
        async def work(client):
            try:
                await client.session.send_request(types.Request(method="tools/call", params={"arguments": {}}), types.CallToolResult)
            except MCPError as error:
                return error
        error = self.run_client(work)
        self.assertIsInstance(error, MCPError)
        self.assertEqual(error.code, types.INVALID_PARAMS)

    def test_client_script_prints_the_recorded_output(self):
        run = subprocess.run([sys.executable, "client.py"], cwd=HERE, capture_output=True, text=True, timeout=60, check=True)
        self.assertEqual(run.stdout, (HERE / "client-output.txt").read_text(encoding="utf-8"))

    def test_saved_calls_match_a_fresh_run(self):
        saved = json.loads((HERE / "recorded-calls.json").read_text(encoding="utf-8"))
        self.assertEqual(asyncio.run(record.collect()), saved)


if __name__ == "__main__":
    unittest.main()

Follow validation and output conversion

Inspect the recorded messages below to distinguish the protocol boundary from the tool result.

The two kinds of failure look different on the wire. A tool error is a normal JSON-RPC result. Its content carries the message, and it has no structuredContent, because there is no Task to return:

{
  "content": [
    {
      "type": "text",
      "text": "Error executing tool complete_task: No task with ID T-9. Call list_tasks to see the valid IDs."
    }
  ],
  "isError": true
}

A protocol error replaces the result with a JSON-RPC error object, here {"code":-32602,"message":"Invalid request parameters"}. The specification says clients should pass tool errors to the model so it can correct itself, while protocol errors are “less likely to result in successful recovery”. Raise ToolError for anything the caller can fix by changing its arguments. Reserve MCPError, which this SDK serializes as a protocol error, for requests the server cannot process at all.

Harden the tools for agent callers

• Validate every input. The specification requires servers to validate all tool inputs. The model writes the arguments, so treat them as untrusted input. Constraints in the type hints run before your code, and the published schema tells the model the same rules, so a schema violation is rejected before tool work begins.
• Validation is not authorization. The same security considerations also require access controls. A valid call is not an allowed call: complete_task accepts any well-formed task ID from any client that can connect, because nothing in this server checks who the caller is or whether they may close that task. That is acceptable for a local stdio demo. A shared server must check the caller’s identity and permissions before a write, as the remote MCP service lesson does with scoped tokens.
• Retries and idempotency. Agents retry calls after timeouts. complete_task is safe to repeat: a second call returns the same done task. add_task is not: a retried call adds a second task with a new ID. For a create tool in a real tracker, accept a client-chosen idempotency key or reject a duplicate open title, and say so in the description.
• Error messages leave the server. Anything in a ToolError message reaches the model and, through it, possibly the user. Keep database errors and internal paths out of it. The SDK already hides unexpected exceptions, and it logs only the names of rejected fields, not the values the caller sent.

A bug inside a tool. The client receives only Error executing tool list_tasks; the traceback stays in the server log. Inspect that log to fix the bug, and keep internal paths and database details out of messages sent to the model.

Check your understanding

Serve MCP over stdio and Streamable HTTP, and test the contracts continues with the next part of the MCP application.

References

Official sources checked October 21, 2026, against mcp 2.3.0 and MCP specification 2026-07-28.

• MCP Python SDK repository and mcp 2.3.0 on PyPI
• MCP specification: tools, including output schemas, annotations and error handling
• Pydantic: fields and constraints