O Server de baixo nível

Tradução automática

Esta página foi traduzida automaticamente a partir da documentação em inglês, e a página em inglês é a versão de referência. Se algo parecer errado, Traduções explica como avisar.

@mcp.tool() é uma camada. Por baixo dela existe uma segunda classe de servidor, Server, que fala MCP cru: você entrega os objetos do protocolo e ela os coloca no fio, sem alterar nada.

O MCPServer é construído em cima dela. Você desce um nível quando a camada de conveniência atrapalha:

Você precisa emitir um schema exato (carregado de um arquivo, gerado a partir de um banco de dados), não um derivado de uma assinatura Python.
Você precisa de controle total do resultado: _meta, is_error, cada chave de structured_content.
Você precisa tratar um método que o MCP não define.

Para todo o resto, fique no MCPServer.

A mesma ferramenta, à mão

Esta é a ferramenta (tool) search_books que Ferramentas escreve em nove linhas de @mcp.tool(), com o açúcar removido:

server.py

from mcp.server import Server, ServerRequestContext
from mcp.types import (
    CallToolRequestParams,
    CallToolResult,
    ListToolsResult,
    PaginatedRequestParams,
    TextContent,
    Tool,
)

SEARCH_BOOKS = Tool(
    name="search_books",
    description="Search the catalog by title or author.",
    input_schema={
        "type": "object",
        "properties": {"query": {"type": "string"}, "limit": {"type": "integer"}},
        "required": ["query", "limit"],
    },
)


async def list_tools(ctx: ServerRequestContext, params: PaginatedRequestParams | None) -> ListToolsResult:
    return ListToolsResult(tools=[SEARCH_BOOKS])


async def call_tool(ctx: ServerRequestContext, params: CallToolRequestParams) -> CallToolResult:
    args = params.arguments or {}
    text = f"Found 3 books matching {args['query']!r} (showing up to {args['limit']})."
    return CallToolResult(content=[TextContent(type="text", text=text)])


server = Server("Bookshop", on_list_tools=list_tools, on_call_tool=call_tool)

Três coisas mudaram, e elas são a API de baixo nível inteira:

Os handlers são parâmetros do construtor. on_list_tools= e on_call_tool= entram em Server(...). Não há decoradores aqui embaixo, e todo handler tem o mesmo formato: async (ctx, params) -> result.
Você escreve o schema de entrada. Tool.input_schema é um dict JSON Schema comum. Ninguém o deriva de anotações de tipo, porque não há anotações de tipo de onde derivar.
Você monta o resultado. CallToolResult(content=[TextContent(...)]), à mão. Nada é encapsulado, convertido ou inferido de uma anotação de retorno.

params é a requisição já parseada: CallToolRequestParams dá .name e .arguments. ctx é um ServerRequestContext: ctx.session para falar de volta com o cliente, ctx.lifespan_context, ctx.request_id e ctx.meta, o _meta de entrada da requisição.

Info

Se você já usou FastAPI, já conhece essa relação. O MCPServer é a camada de decoradores e anotações de tipo; o Server é o Starlette por baixo. Eles não são rivais: o MCPServer constrói um Server e registra nele handlers exatamente como esses.

Experimente

Não existe Inspector para este aqui: mcp dev e mcp run só aceitam um MCPServer. O Client em memória não se importa; ele recebe um Server de baixo nível exatamente como recebe um MCPServer:

main.py

import asyncio

from mcp import Client

from server import server


async def main() -> None:
    async with Client(server) as client:
        result = await client.call_tool("search_books", {"query": "dune", "limit": 5})
        print(result.content)


asyncio.run(main())

[TextContent(type='text', text="Found 3 books matching 'dune' (showing up to 5).", annotations=None, meta=None)]

O mesmo texto que a versão com @mcp.tool() produziu. Duas diferenças honestas:

result.structured_content é None. O servidor de alto nível encapsula um -> str em {"result": ...} para você; aqui ninguém monta o que você não montou.
list_tools retorna o schema que você digitou, caractere por caractere. A versão de alto nível tinha "title": "Query" em cada propriedade e um "title": "search_booksArguments" na raiz: artefatos do Pydantic. Aqui embaixo, se está no fio, foi você quem colocou lá.

Nada é verificado por você

O MCPServer rejeita um argumento ruim antes mesmo de a sua função executar, validando a chamada contra o schema que ele gerou (Ferramentas).

O Server não faz isso. O seu input_schema é anunciado ao cliente; ele nunca é aplicado a params.arguments.

Check

Chame search_books sem limit e o seu args["limit"] levanta KeyError. O cliente vê:

MCPError: Internal server error

Um erro JSON-RPC, código -32603, com uma mensagem deliberadamente genérica: o SDK não vaza o seu traceback para um chamador remoto. O modelo nunca descobre o que fez de errado, então não consegue tentar de novo. (Em um teste, raise_exceptions=True expõe a exceção real; veja Testes.)

Isso se generaliza. Uma exceção levantada de um handler de baixo nível é sempre um erro de protocolo, nunca um resultado de ferramenta com is_error=True. Se você quer que o modelo leia a falha e se recupere, valide params.arguments você mesmo e retorne CallToolResult(content=[TextContent(...)], is_error=True). Os dois tipos de falha são o assunto de Tratando erros.

Duas ferramentas, um handler

on_call_tool é o único ponto de entrada para todas as ferramentas do servidor. Você roteia por params.name:

server.py

from mcp.server import Server, ServerRequestContext
from mcp.types import (
    CallToolRequestParams,
    CallToolResult,
    ListToolsResult,
    PaginatedRequestParams,
    TextContent,
    Tool,
)

SEARCH_BOOKS = Tool(
    name="search_books",
    description="Search the catalog by title or author.",
    input_schema={
        "type": "object",
        "properties": {"query": {"type": "string"}, "limit": {"type": "integer"}},
        "required": ["query", "limit"],
    },
)

ADD_BOOK = Tool(
    name="add_book",
    description="Add a book to the catalog.",
    input_schema={
        "type": "object",
        "properties": {"title": {"type": "string"}, "author": {"type": "string"}, "year": {"type": "integer"}},
        "required": ["title", "author", "year"],
    },
)


async def list_tools(ctx: ServerRequestContext, params: PaginatedRequestParams | None) -> ListToolsResult:
    return ListToolsResult(tools=[SEARCH_BOOKS, ADD_BOOK])


async def call_tool(ctx: ServerRequestContext, params: CallToolRequestParams) -> CallToolResult:
    args = params.arguments or {}
    if params.name == "search_books":
        text = f"Found 3 books matching {args['query']!r} (showing up to {args['limit']})."
    elif params.name == "add_book":
        text = f"Added {args['title']!r} by {args['author']} ({args['year']})."
    else:
        raise ValueError(f"Unknown tool: {params.name}")
    return CallToolResult(content=[TextContent(type="text", text=text)])


server = Server("Bookshop", on_list_tools=list_tools, on_call_tool=call_tool)

list_tools anuncia as duas. call_tool despacha pelo nome.
O ramo else importa: o Server encaminha sem reclamar um tools/call para um nome que você nunca listou direto para o seu handler. Levantar uma exceção ali transforma a chamada no mesmo -32603 de cima.

Saída estruturada, à mão

Declare output_schema na Tool e coloque structured_content no resultado. Os dois são seus:

server.py

from mcp.server import Server, ServerRequestContext
from mcp.types import (
    CallToolRequestParams,
    CallToolResult,
    ListToolsResult,
    PaginatedRequestParams,
    TextContent,
    Tool,
)

SEARCH_BOOKS = Tool(
    name="search_books",
    description="Search the catalog by title or author.",
    input_schema={
        "type": "object",
        "properties": {"query": {"type": "string"}, "limit": {"type": "integer"}},
        "required": ["query", "limit"],
    },
    output_schema={
        "type": "object",
        "properties": {"matches": {"type": "integer"}, "query": {"type": "string"}},
        "required": ["matches", "query"],
    },
)


async def list_tools(ctx: ServerRequestContext, params: PaginatedRequestParams | None) -> ListToolsResult:
    return ListToolsResult(tools=[SEARCH_BOOKS])


async def call_tool(ctx: ServerRequestContext, params: CallToolRequestParams) -> CallToolResult:
    args = params.arguments or {}
    data = {"matches": 3, "query": args["query"]}
    return CallToolResult(
        content=[TextContent(type="text", text=f"Found 3 books matching {args['query']!r}.")],
        structured_content=data,
    )


server = Server("Bookshop", version="2.0.0", on_list_tools=list_tools, on_call_tool=call_tool)

Chame e o resultado carrega as duas representações:

{
  "content": [{"type": "text", "text": "Found 3 books matching 'dune'."}],
  "structuredContent": {"matches": 3, "query": "dune"},
  "isError": false,
  "resultType": "complete",
  "_meta": {"io.modelcontextprotocol/serverInfo": {"name": "Bookshop", "version": "2.0.0"}}
}

O bloco _meta é o carimbo de identidade do servidor: o SDK o adiciona a todo resultado da era 2026, com a version vinda do construtor (um servidor que não define nenhuma reporta uma string vazia). Um servidor que não deve se identificar pode remover a chave com um middleware, que é dono dos resultados que retorna.

O servidor nunca compara os dois campos. O Client deste SDK compara: retorne um structured_content que não satisfaz o output_schema que você declarou e call_tool levanta um RuntimeError que começa com Invalid structured content returned by tool search_books e segue citando a falha do jsonschema. Prometer um schema é barato; cumprir a promessa é com você. A escada inteira de tipos de retorno e schemas está em Saída estruturada.

`_meta`: para a aplicação, não para o modelo

content é a parte da resposta que o modelo lê. structured_content é a mesma resposta como dados tipados. _meta é o terceiro canal: dados que viajam junto com o resultado para a aplicação cliente, sem fazer parte da resposta de forma alguma.

Use para IDs de registro, IDs de trace, qualquer coisa de que a sua UI precisa e o seu prompt não:

server.py

from mcp.server import Server, ServerRequestContext
from mcp.types import (
    CallToolRequestParams,
    CallToolResult,
    ListToolsResult,
    PaginatedRequestParams,
    TextContent,
    Tool,
)

SEARCH_BOOKS = Tool(
    name="search_books",
    description="Search the catalog by title or author.",
    input_schema={
        "type": "object",
        "properties": {"query": {"type": "string"}, "limit": {"type": "integer"}},
        "required": ["query", "limit"],
    },
    output_schema={
        "type": "object",
        "properties": {"matches": {"type": "integer"}, "query": {"type": "string"}},
        "required": ["matches", "query"],
    },
)


async def list_tools(ctx: ServerRequestContext, params: PaginatedRequestParams | None) -> ListToolsResult:
    return ListToolsResult(tools=[SEARCH_BOOKS])


async def call_tool(ctx: ServerRequestContext, params: CallToolRequestParams) -> CallToolResult:
    args = params.arguments or {}
    data = {"matches": 3, "query": args["query"]}
    return CallToolResult(
        content=[TextContent(type="text", text=f"Found 3 books matching {args['query']!r}.")],
        structured_content=data,
        _meta={"bookshop/record_ids": ["bk_17", "bk_42", "bk_99"]},
    )


server = Server("Bookshop", on_list_tools=list_tools, on_call_tool=call_tool)

Você o constrói como _meta=, o nome no fio. O cliente o lê de volta como result.meta.
Use namespace nas suas chaves (bookshop/record_ids). As chaves io.modelcontextprotocol/* são reservadas pelo protocolo.

Warning

_meta é uma convenção entre você e a aplicação cliente, não uma garantia sobre o que chega ao modelo. O host decide o que renderiza. Nunca coloque um segredo em nenhuma parte de um resultado de ferramenta.

As capacidades seguem os seus handlers

Um Server anuncia exatamente as famílias de métodos para as quais você deu handlers. O Bookshop acima passa on_list_tools e on_call_tool e nada mais, então um cliente que se conecta a ele vê:

{"tools": {"listChanged": false}}

Sem resources, sem prompts: não há nada que os sustente. Passe on_list_prompts e prompts aparece; passe on_completion e completions aparece.

O MCPServer sempre anuncia ferramentas, recursos e prompts, tenha você registrado algum ou não, porque os seus managers sempre existem. Aqui embaixo a declaração é a chamada ao construtor.

O genérico do lifespan

O Server é genérico no tipo que o seu lifespan produz. Anote uma vez e o objeto fica tipado em todo lugar onde aparece:

server.py

from collections.abc import AsyncIterator
from contextlib import asynccontextmanager
from dataclasses import dataclass

from mcp.server import Server, ServerRequestContext
from mcp.types import (
    CallToolRequestParams,
    CallToolResult,
    ListToolsResult,
    PaginatedRequestParams,
    TextContent,
    Tool,
)


@dataclass
class Catalog:
    books: list[str]

    def search(self, query: str) -> list[str]:
        return [title for title in self.books if query.lower() in title.lower()]


@asynccontextmanager
async def lifespan(server: Server[Catalog]) -> AsyncIterator[Catalog]:
    yield Catalog(books=["Dune", "Dune Messiah", "Children of Dune"])


SEARCH_BOOKS = Tool(
    name="search_books",
    description="Search the catalog by title or author.",
    input_schema={
        "type": "object",
        "properties": {"query": {"type": "string"}},
        "required": ["query"],
    },
)


async def list_tools(ctx: ServerRequestContext[Catalog], params: PaginatedRequestParams | None) -> ListToolsResult:
    return ListToolsResult(tools=[SEARCH_BOOKS])


async def call_tool(ctx: ServerRequestContext[Catalog], params: CallToolRequestParams) -> CallToolResult:
    matches = ctx.lifespan_context.search((params.arguments or {})["query"])
    text = f"Found {len(matches)} books: {', '.join(matches)}."
    return CallToolResult(content=[TextContent(type="text", text=text)])


server = Server("Bookshop", lifespan=lifespan, on_list_tools=list_tools, on_call_tool=call_tool)

O lifespan é um Callable[[Server[Catalog]], AbstractAsyncContextManager[Catalog]]; @asynccontextmanager em um gerador async dá exatamente isso.
O que quer que ele produza com yield vira ctx.lifespan_context, e como os handlers são anotados com ServerRequestContext[Catalog], .search(...) tem autocompletar e passa na checagem de tipos.
Ele é aberto uma vez quando o servidor inicia e fechado uma vez quando para. Inicialização, encerramento e a versão do MCPServer da mesma ideia estão em Lifespan.

Sem um lifespan=, ctx.lifespan_context é um dict vazio.

Um método só seu

O construtor cobre os métodos que o MCP define. add_request_handler cobre todo o resto:

server.py

from pydantic import BaseModel

from mcp.server import Server, ServerRequestContext
from mcp.types import (
    CallToolRequestParams,
    CallToolResult,
    ListToolsResult,
    PaginatedRequestParams,
    RequestParams,
    TextContent,
    Tool,
)

SEARCH_BOOKS = Tool(
    name="search_books",
    description="Search the catalog by title or author.",
    input_schema={
        "type": "object",
        "properties": {"query": {"type": "string"}, "limit": {"type": "integer"}},
        "required": ["query", "limit"],
    },
)


async def list_tools(ctx: ServerRequestContext, params: PaginatedRequestParams | None) -> ListToolsResult:
    return ListToolsResult(tools=[SEARCH_BOOKS])


async def call_tool(ctx: ServerRequestContext, params: CallToolRequestParams) -> CallToolResult:
    args = params.arguments or {}
    text = f"Found 3 books matching {args['query']!r} (showing up to {args['limit']})."
    return CallToolResult(content=[TextContent(type="text", text=text)])


class ReindexParams(RequestParams):
    full: bool = False


class ReindexResult(BaseModel):
    indexed: int


async def reindex(ctx: ServerRequestContext, params: ReindexParams) -> ReindexResult:
    return ReindexResult(indexed=3)


server = Server("Bookshop", on_list_tools=list_tools, on_call_tool=call_tool)
server.add_request_handler("bookshop/reindex", ReindexParams, reindex)

O primeiro argumento é a string do método. Notificações têm um irmão gêmeo, add_notification_handler.
params_type é o modelo contra o qual os params recebidos são validados antes de o seu handler executar, então métodos personalizados recebem a validação que as ferramentas não recebem. Faça subclasse de RequestParams para que o campo _meta seja parseado como o de qualquer outro método.
O handler retorna um BaseModel, um dict ou None. O SDK serializa isso no resultado JSON-RPC.

Uma ressalva honesta: o Client de alto nível só tem verbos para os métodos que o MCP define, então não existe client.reindex(). Um método de fornecedor é para um par que já sabe que ele existe: um cliente que você também distribui, ou outro serviço seu falando JSON-RPC.

Um método que você não pode reivindicar:

ValueError: 'initialize' is handled by the server runner and cannot be overridden;
use Server.middleware to observe or wrap initialization

O handshake pertence ao runner. server/discover, ping e todos os outros embutidos são seus para substituir.

Tip

Server.middleware, mencionado naquele erro, envolve toda mensagem de entrada, inclusive initialize. Se o que você quer é observar ou reescrever o tráfego em vez de responder a um método novo, comece por Middleware.

Os outros handlers

Cada um destes é uma ideia para a qual você já tem o vocabulário; cada um tem sua própria página.

on_call_tool, on_get_prompt e on_read_resource podem retornar um InputRequiredResult em vez do resultado normal para pausar a chamada e pedir entrada ao cliente; veja Requisições de múltiplas idas e voltas. Fiel a este nível, nada é instalado para você: enquanto o MCPServer sela o requestState por padrão, aqui o request_state que você define atravessa o fio exatamente como foi escrito até você optar com server.middleware.append(RequestStateBoundary(RequestStateSecurity(keys=[...]), default_audience=server.name)): uma linha (os dois nomes são importados de mcp.server.request_state) para a mesma selagem e verificação que o MCPServer faz (Protegendo o requestState).
on_list_resources, on_read_resource, on_list_prompts, on_get_prompt, on_completion têm o mesmo formato (ctx, params) -> result para as outras primitivas.
on_subscriptions_listen serve o stream subscriptions/listen de 2026-07-28. Passe um ListenHandler construído sobre um SubscriptionBus e publique eventos no bus a partir dos seus outros handlers; veja Assinaturas para a composição completa.
server.streamable_http_app() retorna o mesmo app Starlette que o do MCPServer; faça o deploy dele do jeito que Executando o seu servidor faz o deploy de qualquer outro app ASGI. Não existe server.run(transport=...) aqui embaixo: server.run(read_stream, write_stream, server.create_initialization_options()) conduz uma conexão sobre um par de streams, e essa única linha é a história completa.

Recapitulando

O Server de baixo nível recebe os seus handlers como parâmetros do construtor on_*; todo handler é async (ctx, params) -> result.
Você escreve o dict input_schema e você monta o CallToolResult. Nada é derivado, encapsulado ou validado por você.
Uma exceção em um handler é um erro de protocolo -32603. Um erro de ferramenta que o modelo consegue ler é um CallToolResult com is_error=True que você retorna.
O _meta no resultado é endereçado à aplicação cliente, não ao modelo.
Server[T] é genérico no que o seu lifespan produz; ctx.lifespan_context é um T tipado.
add_request_handler(method, params_type, handler) serve qualquer método. initialize é reservado.
As capacidades que um Server anuncia são derivadas de quais handlers você registrou.

Client(server) tratou os dois servidores de forma idêntica porque eles são o mesmo protocolo, e essa é justamente a ideia. A próxima camada abaixo nem é uma classe: é Middleware.

O Server de baixo nível

A mesma ferramenta, à mão

Experimente

Nada é verificado por você

Duas ferramentas, um handler

Saída estruturada, à mão

_meta: para a aplicação, não para o modelo

As capacidades seguem os seus handlers

O genérico do lifespan

Um método só seu

Os outros handlers

Recapitulando

`_meta`: para a aplicação, não para o modelo