Extensões

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.

Uma extensão é um pacote opcional de comportamento MCP reunido sob um único identificador.

Em um servidor, ela pode contribuir com ferramentas (tools), recursos e novos métodos de requisição, e pode envolver tools/call. Em um cliente, ela pode reivindicar formatos extras de resultado de tools/call e observar notificações de fornecedores. Cada lado se anuncia no seu próprio capabilities.extensions, e nada muda para quem não pediu nada. Esse é o contrato (SEP-2133), e ele tem uma regra de ouro: extensões vêm desligadas por padrão.

Usando uma extensão

Passe as instâncias na construção:

server.py

from mcp.server.apps import Apps
from mcp.server.mcpserver import MCPServer

mcp = MCPServer("demo", extensions=[Apps()])

Pronto. O servidor agora anuncia io.modelcontextprotocol/ui em capabilities.extensions e serve tudo o que a extensão contribui.

Apps é a extensão de referência embutida, e ela tem uma página própria: MCP Apps.

Note

As extensões são fixadas na construção. Não existe um add_extension para chamar depois: o mapa de capacidades de um servidor não deve mudar enquanto há clientes conectados a ele.

O mapa de capacidades viaja em server/discover, que é um caminho da 2026-07-28. Um handshake initialize legado não tem onde colocá-lo, então um cliente legado simplesmente não enxerga a extensão. Projete pensando nisso: uma extensão amplia um servidor, ela não pode ser a única forma de usá-lo.

Escrevendo a sua

Herde de Extension e sobrescreva apenas o que precisar. Todo método tem um padrão.

O identificador

from mcp.server.extension import Extension


class Stamps(Extension):
    identifier = "com.example/stamps"

O identificador é uma string vendor-prefix/name que segue a gramática de chaves _meta da especificação: rótulos separados por ponto (cada um começa com uma letra e termina com uma letra ou dígito), uma barra e então o nome. Ele é validado quando a classe é definida, então um erro de digitação não espera o servidor subir:

TypeError: Stamps.identifier must be a `vendor-prefix/name` string
(reverse-DNS prefix required), got 'stamps'

Use como prefixo um domínio que você controla. io.modelcontextprotocol/* é reservado para extensões especificadas pelo próprio projeto MCP.

Contribuindo com ferramentas

A menor extensão útil é uma ferramenta e um mapa de configurações:

server.py

from collections.abc import Sequence
from typing import Any

from mcp import Client
from mcp.server.extension import Extension, ToolBinding
from mcp.server.mcpserver import MCPServer


def stamp(text: str) -> str:
    """Stamp a message with the office seal."""
    return f"[stamped] {text}"


class Stamps(Extension):
    """A purely additive extension: one tool, one capability entry."""

    identifier = "com.example/stamps"

    def settings(self) -> dict[str, Any]:
        return {"sealed": True}

    def tools(self) -> Sequence[ToolBinding]:
        return [ToolBinding(fn=stamp)]


mcp = MCPServer("post-office", extensions=[Stamps()])


async def main() -> None:
    async with Client(mcp) as client:
        print(client.server_capabilities.extensions)
        # {'com.example/stamps': {'sealed': True}}
        result = await client.call_tool("stamp", {"text": "hello"})
        print(result.content)
        # [TextContent(text='[stamped] hello')]

tools() retorna ToolBindings. O servidor registra cada uma exatamente como se você tivesse chamado mcp.add_tool(...) por conta própria: mesma geração de schema, mesma injeção de Context, tudo igual.
settings() é o valor anunciado em capabilities.extensions["com.example/stamps"]. Retorne {} (o padrão) para anunciar a extensão sem configurações.
A extensão nunca recebe o servidor. Ela declara contribuições como dados; o MCPServer as consome. Não existe um self.server para modificar.

E main() é a prova, um cliente em memória direto contra mcp:

server.py

from collections.abc import Sequence
from typing import Any

from mcp import Client
from mcp.server.extension import Extension, ToolBinding
from mcp.server.mcpserver import MCPServer


def stamp(text: str) -> str:
    """Stamp a message with the office seal."""
    return f"[stamped] {text}"


class Stamps(Extension):
    """A purely additive extension: one tool, one capability entry."""

    identifier = "com.example/stamps"

    def settings(self) -> dict[str, Any]:
        return {"sealed": True}

    def tools(self) -> Sequence[ToolBinding]:
        return [ToolBinding(fn=stamp)]


mcp = MCPServer("post-office", extensions=[Stamps()])


async def main() -> None:
    async with Client(mcp) as client:
        print(client.server_capabilities.extensions)
        # {'com.example/stamps': {'sealed': True}}
        result = await client.call_tool("stamp", {"text": "hello"})
        print(result.content)
        # [TextContent(text='[stamped] hello')]

Servindo seus próprios métodos

Uma extensão pode registrar novos métodos de requisição: seus próprios verbos, servidos ao lado dos da especificação:

server.py

from collections.abc import Sequence
from typing import Any, Literal

from pydantic import Field

import mcp.types as types
from mcp import Client
from mcp.client import advertise
from mcp.server.context import ServerRequestContext
from mcp.server.extension import Extension, MethodBinding
from mcp.server.mcpserver import MCPServer, require_client_extension

EXTENSION_ID = "com.example/search"


class SearchParams(types.RequestParams):
    query: str
    limit: int = Field(default=10, ge=1, le=100)


class SearchResult(types.Result):
    items: list[str]


class SearchRequest(types.Request[SearchParams, Literal["com.example/search"]]):
    method: Literal["com.example/search"] = "com.example/search"
    params: SearchParams


async def search(ctx: ServerRequestContext[Any, Any], params: SearchParams) -> SearchResult:
    require_client_extension(ctx, EXTENSION_ID)
    return SearchResult(items=[f"{params.query}-{n}" for n in range(params.limit)])


class Search(Extension):
    """An extension that serves its own request method."""

    identifier = EXTENSION_ID

    def methods(self) -> Sequence[MethodBinding]:
        return [
            MethodBinding(
                "com.example/search",
                SearchParams,
                search,
                protocol_versions=frozenset({"2026-07-28"}),
            )
        ]


mcp = MCPServer("catalog", extensions=[Search()])


async def main() -> None:
    async with Client(mcp, extensions=[advertise(EXTENSION_ID)]) as client:
        request = SearchRequest(params=SearchParams(query="mcp", limit=3))
        result = await client.session.send_request(request, SearchResult)
        print(result.items)
        # ['mcp-0', 'mcp-1', 'mcp-2']

SearchParams herda de RequestParams, então o envelope _meta de 2026 é analisado de forma uniforme e seu handler recebe parâmetros validados, nunca um dict cru. Limite o que o cliente controla: Field(ge=1, le=100) rejeita um limit absurdo antes que seu código aloque qualquer coisa para ele.
require_client_extension(ctx, EXTENSION_ID) é a barreira: um cliente que não declarou a extensão recebe o erro -32021 (capacidade obrigatória do cliente ausente), com o payload requiredCapabilities legível por máquina que a especificação pede.
protocol_versions=frozenset({"2026-07-28"}) fixa o método em uma única versão de protocolo. Em qualquer outra versão o cliente recebe METHOD_NOT_FOUND, exatamente como se o método não existisse ali. Para esse cliente, não existe.

Os métodos são estritamente aditivos. O SDK impõe isso na construção, não em tempo de execução:

Um MethodBinding para um método definido pela especificação (tools/list, completion/complete, ...) lança ValueError quando o binding é construído. Os verbos centrais pertencem ao servidor.
Duas extensões vinculando o mesmo método lançam quando a segunda se registra. A última escrita vencer é como plugins corrompem uns aos outros; não fazemos isso.
Um conjunto protocol_versions vazio também lança: um método que nunca pode ser servido é um bug, não uma configuração.

O lado do cliente

O main() do mesmo arquivo é a história inteira do cliente, com as duas metades:

server.py

from collections.abc import Sequence
from typing import Any, Literal

from pydantic import Field

import mcp.types as types
from mcp import Client
from mcp.client import advertise
from mcp.server.context import ServerRequestContext
from mcp.server.extension import Extension, MethodBinding
from mcp.server.mcpserver import MCPServer, require_client_extension

EXTENSION_ID = "com.example/search"


class SearchParams(types.RequestParams):
    query: str
    limit: int = Field(default=10, ge=1, le=100)


class SearchResult(types.Result):
    items: list[str]


class SearchRequest(types.Request[SearchParams, Literal["com.example/search"]]):
    method: Literal["com.example/search"] = "com.example/search"
    params: SearchParams


async def search(ctx: ServerRequestContext[Any, Any], params: SearchParams) -> SearchResult:
    require_client_extension(ctx, EXTENSION_ID)
    return SearchResult(items=[f"{params.query}-{n}" for n in range(params.limit)])


class Search(Extension):
    """An extension that serves its own request method."""

    identifier = EXTENSION_ID

    def methods(self) -> Sequence[MethodBinding]:
        return [
            MethodBinding(
                "com.example/search",
                SearchParams,
                search,
                protocol_versions=frozenset({"2026-07-28"}),
            )
        ]


mcp = MCPServer("catalog", extensions=[Search()])


async def main() -> None:
    async with Client(mcp, extensions=[advertise(EXTENSION_ID)]) as client:
        request = SearchRequest(params=SearchParams(query="mcp", limit=3))
        result = await client.session.send_request(request, SearchResult)
        print(result.items)
        # ['mcp-0', 'mcp-1', 'mcp-2']

Client(..., extensions=[advertise(EXTENSION_ID)]) declara a extensão. As declarações viram ClientCapabilities.extensions: em uma conexão 2026-07-28 o mapa viaja no envelope _meta de cada requisição, então o servidor o vê em toda requisição; em uma conexão legada ele vai no handshake initialize. O código do servidor não se importa com qual: require_client_extension(ctx, ...) e ctx.session.check_client_capability(...) leem a fonte certa nos dois caminhos.
Métodos de fornecedor descem uma camada para client.session.send_request(...); Client só ganha métodos de primeira classe para verbos da especificação. send_request aceita qualquer subclasse de Request, então a requisição do fornecedor passa como está.

Interceptando `tools/call`

O único hook interceptador. Sobrescreva intercept_tool_call para observar, curto-circuitar ou vetar uma chamada de ferramenta:

server.py

import logging
from typing import Any

from mcp.server.context import CallNext, HandlerResult, ServerRequestContext
from mcp.server.extension import Extension
from mcp.server.mcpserver import MCPServer
from mcp.types import CallToolRequestParams

logger = logging.getLogger(__name__)


class AuditLog(Extension):
    """Observe every tools/call without touching its result."""

    identifier = "com.example/audit"

    async def intercept_tool_call(
        self,
        params: CallToolRequestParams,
        ctx: ServerRequestContext[Any, Any],
        call_next: CallNext,
    ) -> HandlerResult:
        logger.info("tool %r called", params.name)
        return await call_next(ctx)


mcp = MCPServer("audited", extensions=[AuditLog()])


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

params é o CallToolRequestParams validado: você recebe params.name e params.arguments sem tocar em JSON cru. É também o que decide qual chamada de ferramenta é executada: passar um contexto reescrito por call_next muda o que o handler observa em ctx, não a invocação da ferramenta. Reescrita de requisição no nível do protocolo pertence ao Middleware.
call_next(ctx) executa o resto da cadeia e retorna o resultado do handler. Retorne-o sem alterações (observar), retorne outra coisa (substituir) ou lance um MCPError (recusar). O que você retornar é serializado como qualquer resultado de handler, incluindo o carimbo de identidade serverInfo da era 2026, então um interceptador que curto-circuita nunca produz uma resposta anônima ou fora do schema.
Com várias extensões, os interceptadores se aninham na ordem de registro: a primeira extensão em extensions=[...] é a mais externa.
A implementação padrão é um repasse direto, e um servidor cujas extensões nunca sobrescrevem esse hook mantém o handler puro de tools/call intocado. Você não paga pelo que não usa.

O hook envolve tools/call e nada mais. Para preocupações que valem para toda mensagem, use o Middleware. É para isso que ele serve.

Usando uma extensão de cliente

Uma extensão de cliente é o mesmo contrato visto do lado consumidor: um pacote de comportamento do lado do cliente reunido sob um único identificador. Passe as instâncias para Client(extensions=[...]) e chame as ferramentas normalmente:

client.py

from collections.abc import Sequence
from typing import Any, Literal

import mcp.types as types
from mcp import Client
from mcp.client import ClaimContext, ClientExtension, ResultClaim
from mcp.server.context import CallNext, HandlerResult, ServerRequestContext
from mcp.server.extension import Extension
from mcp.server.mcpserver import MCPServer, require_client_extension

EXTENSION_ID = "com.example/receipts"


class ReceiptResult(types.Result):
    """The claimed result shape; `result_type` pins the wire tag."""

    result_type: Literal["receipt"] = "receipt"
    receipt_token: str


class ReceiptIssuer(Extension):
    """Server half: answers `buy` with a receipt instead of a final result."""

    identifier = EXTENSION_ID

    async def intercept_tool_call(
        self,
        params: types.CallToolRequestParams,
        ctx: ServerRequestContext[Any, Any],
        call_next: CallNext,
    ) -> HandlerResult:
        if params.name != "buy":
            return await call_next(ctx)
        require_client_extension(ctx, EXTENSION_ID)
        return {"resultType": "receipt", "receiptToken": "r-117"}


class Receipts(ClientExtension):
    """Client half: claims the `receipt` shape and supplies the code that finishes it."""

    identifier = EXTENSION_ID

    def claims(self) -> Sequence[ResultClaim[Any]]:
        return [ResultClaim(result_type="receipt", model=ReceiptResult, resolve=self._redeem)]

    async def _redeem(self, claimed: ReceiptResult, ctx: ClaimContext) -> types.CallToolResult:
        return await ctx.session.call_tool("redeem", {"token": claimed.receipt_token})


mcp = MCPServer("shop", extensions=[ReceiptIssuer()])


@mcp.tool()
def buy(item: str) -> types.CallToolResult:
    """Buy an item."""
    raise NotImplementedError  # ReceiptIssuer answers `buy` before the tool runs


@mcp.tool()
def redeem(token: str) -> str:
    """Exchange a receipt token for the goods."""
    return f"goods for {token}"


async def main() -> None:
    async with Client(mcp, extensions=[Receipts()]) as client:
        result = await client.call_tool("buy", {"item": "lamp"})
        print(result.content)
        # [TextContent(text='goods for r-117')]

call_tool("buy", ...) retorna um CallToolResult comum, como toda outra chamada. O que a extensão mudou: o servidor agora pode responder a buy com um formato de resultado receipt em vez de um resultado final, e Receipts o finaliza (aqui, resgatando o recibo com uma chamada seguinte) antes de call_tool retornar. Nada muda no ponto da chamada.

Tire a extensão e nada disso existe: a barreira do servidor recusa um cliente que não a declarou (erro -32021), e um formato reivindicado vindo de um servidor que pula a barreira falha na validação, exatamente como a especificação exige para um resultType não reconhecido. Desligado por padrão, nas duas pontas da conexão.

Para anunciar um identificador sem nenhum comportamento do lado do cliente (o servidor faz a barreira pela capacidade, o cliente não faz nada, como no cliente de busca acima), use advertise():

from mcp.client import advertise

client = Client(mcp, extensions=[advertise("com.example/search")])

Escrevendo uma extensão de cliente

Herde de ClientExtension e sobrescreva apenas o que precisar. Três tipos de contribuição, cada um com um padrão: settings(), claims() e notifications().

client.py

from collections.abc import Sequence
from typing import Any, Literal

import mcp.types as types
from mcp import Client
from mcp.client import ClaimContext, ClientExtension, ResultClaim
from mcp.server.context import CallNext, HandlerResult, ServerRequestContext
from mcp.server.extension import Extension
from mcp.server.mcpserver import MCPServer, require_client_extension

EXTENSION_ID = "com.example/receipts"


class ReceiptResult(types.Result):
    """The claimed result shape; `result_type` pins the wire tag."""

    result_type: Literal["receipt"] = "receipt"
    receipt_token: str


class ReceiptIssuer(Extension):
    """Server half: answers `buy` with a receipt instead of a final result."""

    identifier = EXTENSION_ID

    async def intercept_tool_call(
        self,
        params: types.CallToolRequestParams,
        ctx: ServerRequestContext[Any, Any],
        call_next: CallNext,
    ) -> HandlerResult:
        if params.name != "buy":
            return await call_next(ctx)
        require_client_extension(ctx, EXTENSION_ID)
        return {"resultType": "receipt", "receiptToken": "r-117"}


class Receipts(ClientExtension):
    """Client half: claims the `receipt` shape and supplies the code that finishes it."""

    identifier = EXTENSION_ID

    def claims(self) -> Sequence[ResultClaim[Any]]:
        return [ResultClaim(result_type="receipt", model=ReceiptResult, resolve=self._redeem)]

    async def _redeem(self, claimed: ReceiptResult, ctx: ClaimContext) -> types.CallToolResult:
        return await ctx.session.call_tool("redeem", {"token": claimed.receipt_token})


mcp = MCPServer("shop", extensions=[ReceiptIssuer()])


@mcp.tool()
def buy(item: str) -> types.CallToolResult:
    """Buy an item."""
    raise NotImplementedError  # ReceiptIssuer answers `buy` before the tool runs


@mcp.tool()
def redeem(token: str) -> str:
    """Exchange a receipt token for the goods."""
    return f"goods for {token}"


async def main() -> None:
    async with Client(mcp, extensions=[Receipts()]) as client:
        result = await client.call_tool("buy", {"item": "lamp"})
        print(result.content)
        # [TextContent(text='goods for r-117')]

O identificador segue a mesma gramática do servidor, validada quando a classe é definida.
claims() retorna ResultClaims: uma tag de protocolo, o modelo que a analisa e o resolvedor que a finaliza. O modelo precisa fixar a tag com result_type: Literal["receipt"] e não pode herdar dos tipos de resultado centrais do verbo; as duas coisas são impostas quando a claim é construída. Campos de fornecedor como receipt_token viajam pela conexão como estão: um formato substituído chega ao cliente literalmente.
O resolvedor recebe o modelo analisado e um ClaimContext; ctx.session é o mesmo handle público que client.session, então as chamadas seguintes são chamadas comuns de sessão. Ele retorna o CallToolResult normal do verbo.
settings() é o valor anunciado em ClientCapabilities.extensions[identifier], lido uma vez na construção do Client.

notifications() declara notificações de servidor de fornecedor a observar:

def notifications(self) -> Sequence[NotificationBinding[Any]]:
    return [NotificationBinding(method="notifications/receipts", params_type=ReceiptEvent, handler=self.on_receipt)]

O handler recebe parâmetros validados um de cada vez, na ordem de despacho. Ele observa; não pode vetar nem responder.

Duas regras discretas. As claims ficam ativas apenas em conexões 2026-07-28, e o anúncio de capacidade as acompanha: em uma conexão legada as claims se dissolvem e o identificador sai do anúncio junto com elas, então o cliente nunca anuncia uma extensão cujos formatos ele rejeitaria. E quando você mesmo quer o formato reivindicado em vez do resolvedor, chame client.session.call_tool(..., allow_claimed=True); sem essa flag, um formato reivindicado que chega a um chamador no nível da sessão lança UnexpectedClaimedResult.

Verbos de extensão

Os métodos de requisição próprios de uma extensão não precisam de registro no lado do cliente. Um tipo de requisição de fornecedor herda de mcp.types.Request e passa por client.session.send_request, como em Servindo seus próprios métodos. Um acréscimo: quando uma chave de params precisa viajar no header Mcp-Name (especificações de extensão como tasks exigem isso para seus verbos), o tipo de requisição declara name_param:

client.py

from collections.abc import Sequence
from typing import Any, Literal

import mcp.types as types
from mcp import Client
from mcp.client import advertise
from mcp.server.context import ServerRequestContext
from mcp.server.extension import Extension, MethodBinding
from mcp.server.mcpserver import MCPServer

EXTENSION_ID = "com.example/jobs"


class JobParams(types.RequestParams):
    job_id: str


class JobStatus(types.Result):
    status: str


class JobStatusRequest(types.Request[JobParams, Literal["com.example/jobs.status"]]):
    method: Literal["com.example/jobs.status"] = "com.example/jobs.status"
    params: JobParams
    name_param = "jobId"  # params["jobId"] rides the Mcp-Name header


async def job_status(ctx: ServerRequestContext[Any, Any], params: JobParams) -> JobStatus:
    return JobStatus(status=f"{params.job_id} is running")


class Jobs(Extension):
    """An extension whose verb names its subject, so the header can route on it."""

    identifier = EXTENSION_ID

    def methods(self) -> Sequence[MethodBinding]:
        return [MethodBinding("com.example/jobs.status", JobParams, job_status)]


mcp = MCPServer("worker", extensions=[Jobs()])


async def main() -> None:
    async with Client(mcp, extensions=[advertise(EXTENSION_ID)]) as client:
        request = JobStatusRequest(params=JobParams(job_id="job-7"))
        result = await client.session.send_request(request, JobStatus)
        print(result.status)
        # job-7 is running

A sessão espelha params["jobId"] em Mcp-Name em todo caminho de envio, e um valor ausente falha de forma explícita em vez de omitir silenciosamente um header obrigatório.

O que uma extensão não pode fazer

A superfície de contribuição é fechada de propósito. No servidor: configurações, ferramentas, recursos, métodos, um interceptador de tools/call. No cliente: configurações, claims de resultado, bindings de notificação. Uma extensão não pode:

Alcançar o host. Ela declara dados; não guarda nenhuma referência ao servidor nem ao cliente.
Substituir comportamento central. Métodos da especificação e tags de resultado centrais são rejeitados na construção (initialize é reservado pelo runner sem exceção); já um binding de notificação encoberto pelo vocabulário central fica em silêncio com um aviso.
Registrar-se depois. Depois que MCPServer(...) ou Client(...) retorna, o conjunto de extensões é o que é.

Se você está brigando com essas paredes, não está escrevendo uma extensão. Está escrevendo um fork. As paredes são a funcionalidade: um usuário que lê extensions=[Apps(), Stamps()] sabe tudo o que essas duas podem ter tocado.

Extensões

Usando uma extensão

Escrevendo a sua

O identificador

Contribuindo com ferramentas

Servindo seus próprios métodos

O lado do cliente

Interceptando tools/call

Usando uma extensão de cliente

Escrevendo uma extensão de cliente

Verbos de extensão

O que uma extensão não pode fazer

Interceptando `tools/call`