Расширения

Машинный перевод

Эта страница переведена с английской документации автоматически, и основной версией остаётся английская страница. Если что-то читается неправильно, на странице Переводы объясняется, как об этом сообщить.

Расширение — это набор поведения MCP, который включается только по желанию и объединён одним идентификатором.

На сервере оно может добавлять инструменты, ресурсы и новые методы запросов, а также оборачивать tools/call. На клиенте — заявлять дополнительные формы результата tools/call и наблюдать за вендорными уведомлениями. Каждая сторона объявляет его в собственном capabilities.extensions, и для тех, кто об этом не просил, ничего не меняется. Таков контракт (SEP-2133), и у него одно золотое правило: по умолчанию расширения выключены.

Использование расширения

Передайте экземпляры при создании:

server.py

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

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

Готово. Теперь сервер объявляет io.modelcontextprotocol/ui в capabilities.extensions и обслуживает всё, что добавляет расширение.

Apps — встроенное эталонное расширение, и ему посвящена отдельная страница: MCP Apps.

Note

Расширения фиксируются при создании. Метода add_extension, который можно вызвать позже, нет: карта возможностей сервера не должна меняться, пока к нему подключены клиенты.

Карта возможностей передаётся через server/discover, а это путь версии 2026-07-28. В рукопожатии initialize старого поколения для неё просто нет места, поэтому клиент старого поколения расширения не видит. Учитывайте это при проектировании: расширение дополняет сервер и не должно быть единственным способом им пользоваться.

Написание собственного расширения

Унаследуйтесь от Extension и переопределите только то, что нужно. У каждого метода есть реализация по умолчанию.

Идентификатор

from mcp.server.extension import Extension


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

Идентификатор — это строка вида vendor-prefix/name, подчиняющаяся грамматике ключей _meta из спецификации: метки, разделённые точками (каждая начинается с буквы и заканчивается буквой или цифрой), косая черта, затем имя. Он проверяется в момент определения класса, так что опечатка не ждёт запуска сервера:

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

В качестве префикса используйте домен, которым вы управляете. io.modelcontextprotocol/* отведён для расширений, описанных самим проектом 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')]

tools() возвращает объекты ToolBinding. Сервер регистрирует каждый из них ровно так же, как если бы вы сами вызвали mcp.add_tool(...): та же генерация схемы, то же внедрение Context, всё то же самое.
settings() — значение, объявляемое в capabilities.extensions["com.example/stamps"]. Верните {} (значение по умолчанию), чтобы объявить расширение без настроек.
Расширение никогда не получает сервер. Оно описывает свой вклад как данные; MCPServer их потребляет. Никакого self.server, который можно было бы менять, нет.

А main() служит доказательством: клиент в памяти, подключённый напрямую к 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')]

Обслуживание собственных методов

Расширение может регистрировать новые методы запросов — собственные глаголы, обслуживаемые рядом с методами спецификации:

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 наследуется от RequestParams, поэтому конверт _meta версии 2026 разбирается единообразно, а обработчик получает провалидированные параметры, а не сырой словарь. Ограничивайте то, чем управляет клиент: Field(ge=1, le=100) отклонит абсурдный limit раньше, чем ваш код что-либо под него выделит.
require_client_extension(ctx, EXTENSION_ID) — это пропускной пункт: клиент, не объявивший расширение, получает ошибку -32021 (отсутствует обязательная возможность клиента) с машиночитаемой полезной нагрузкой requiredCapabilities, которую требует спецификация.
protocol_versions=frozenset({"2026-07-28"}) привязывает метод к одной версии протокола. На любой другой версии клиент получает METHOD_NOT_FOUND — ровно так, как если бы метода там не существовало. Для этого клиента его и нет.

Методы строго аддитивны. SDK проверяет это при создании, а не во время выполнения:

MethodBinding для метода, определённого спецификацией (tools/list, completion/complete, ...), выбрасывает ValueError при создании привязки. Базовые глаголы принадлежат серверу.
Если два расширения привязывают один и тот же метод, исключение выбрасывается при регистрации второго. Принцип «побеждает последняя запись» — это то, как плагины портят друг друга; мы так не делаем.
Пустое множество protocol_versions тоже приводит к исключению: метод, который никогда нельзя обслужить, — это ошибка, а не конфигурация.

Клиентская сторона

main() из того же файла — это вся клиентская часть, обе её половины:

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)]) объявляет расширение. Объявления превращаются в ClientCapabilities.extensions: на подключении версии 2026-07-28 карта передаётся в конверте _meta каждого запроса, так что сервер видит её в каждом запросе; на подключении старого поколения она едет в рукопожатии initialize. Серверному коду всё равно, какой из вариантов: require_client_extension(ctx, ...) и ctx.session.check_client_capability(...) читают нужный источник на обоих путях.
Вендорные методы спускаются на уровень ниже, к client.session.send_request(...); Client обзаводится полноценными методами только для глаголов спецификации. send_request принимает любой подкласс Request, так что вендорный запрос проходит как есть.

Перехват `tools/call`

Единственный перехватывающий хук. Переопределите intercept_tool_call, чтобы наблюдать за вызовом инструмента, завершать его досрочно или запрещать:

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 — провалидированный CallToolRequestParams: params.name и params.arguments доступны без работы с сырым JSON. Он же определяет, какой вызов инструмента выполняется: передача переписанного контекста через call_next меняет то, что обработчик видит в ctx, но не сам вызов инструмента. Переписывание запросов на уровне протокола — задача Middleware.
call_next(ctx) выполняет остаток цепочки и возвращает результат обработчика. Верните его без изменений (наблюдение), верните что-то другое (замена) или выбросьте MCPError (отказ). Всё, что вы вернёте, сериализуется как любой результат обработчика, включая штамп идентичности serverInfo поколения 2026, так что перехватчик, завершающий вызов досрочно, никогда не выдаёт анонимный или не соответствующий схеме ответ.
При нескольких расширениях перехватчики вкладываются друг в друга в порядке регистрации: первое расширение в extensions=[...] — самое внешнее.
Реализация по умолчанию просто пропускает вызов дальше, и сервер, расширения которого не переопределяют этот хук, сохраняет голый обработчик tools/call нетронутым. За то, чем не пользуетесь, платить не приходится.

Хук оборачивает tools/call и ничего больше. Для задач, касающихся каждого сообщения, используйте Middleware. Оно для этого и предназначено.

Использование клиентского расширения

Клиентское расширение — тот же контракт со стороны потребителя: набор клиентского поведения за одним идентификатором. Передайте экземпляры в Client(extensions=[...]) и вызывайте инструменты как обычно:

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", ...) возвращает обычный CallToolResult, как и любой другой вызов. Что изменило расширение: теперь сервер может ответить на buy формой результата receipt вместо окончательного результата, а Receipts доводит её до конца (здесь — погашая квитанцию дополнительным вызовом) до того, как call_tool вернёт управление. В месте вызова не меняется ничего.

Уберите расширение — и ничего этого не будет: пропускной пункт сервера отклонит клиент, который его не объявил (ошибка -32021), а заявленная форма от сервера, пропускающего эту проверку, не пройдёт валидацию — ровно так, как спецификация требует для нераспознанного resultType. Выключено по умолчанию, на обоих концах соединения.

Чтобы объявить идентификатор без какого-либо клиентского поведения (сервер проверяет наличие возможности, клиент ничего не делает — как в клиенте поиска выше), используйте advertise():

from mcp.client import advertise

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

Написание клиентского расширения

Унаследуйтесь от ClientExtension и переопределите только то, что нужно. Три вида вклада, у каждого реализация по умолчанию: settings(), claims() и 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')]

Идентификатор подчиняется той же грамматике, что и на сервере, и проверяется при определении класса.
claims() возвращает объекты ResultClaim: тег в передаваемых данных, модель, которая его разбирает, и резолвер, который доводит результат до конца. Модель обязана зафиксировать тег через result_type: Literal["receipt"] и не должна наследоваться от базовых типов результата этого глагола; и то и другое проверяется при создании заявки. Вендорные поля вроде receipt_token передаются по сети как есть: подставленная форма доходит до клиента дословно.
Резолвер получает разобранную модель и ClaimContext; ctx.session — тот же публичный дескриптор, что и client.session, так что последующие вызовы — это обычные вызовы сессии. Возвращает он обычный для глагола CallToolResult.
settings() — значение, объявляемое в ClientCapabilities.extensions[identifier]; оно считывается один раз при создании Client.

notifications() объявляет вендорные уведомления сервера, за которыми нужно наблюдать:

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

Обработчик получает провалидированные параметры по одному, в порядке диспетчеризации. Он наблюдает; запретить или ответить он не может.

Два негромких правила. Заявки действуют только на подключениях версии 2026-07-28, и объявление возможностей следует за ними: на подключении старого поколения заявки исчезают, а вместе с ними из объявления выпадает и идентификатор, так что клиент никогда не объявляет расширение, формы которого он бы отклонил. А если заявленная форма нужна вам самим, а не резолверу, вызывайте client.session.call_tool(..., allow_claimed=True); без этого флага заявленная форма, дошедшая до вызывающего кода на уровне сессии, приводит к исключению UnexpectedClaimedResult.

Глаголы расширения

Собственные методы запросов расширения не требуют регистрации на стороне клиента. Тип вендорного запроса наследуется от mcp.types.Request и отправляется через client.session.send_request, как в разделе Обслуживание собственных методов. Одно дополнение: когда ключ из params должен передаваться в заголовке Mcp-Name (спецификации расширений, например tasks, требуют этого для своих глаголов), тип запроса объявляет 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

Сессия дублирует params["jobId"] в Mcp-Name на каждом пути отправки, а отсутствующее значение приводит к явной ошибке, а не к молчаливому пропуску обязательного заголовка.

Чего расширение не может

Поверхность вклада закрыта намеренно. На сервере: настройки, инструменты, ресурсы, методы, один перехватчик tools/call. На клиенте: настройки, заявки на результаты, привязки уведомлений. Расширение не может:

Дотянуться до хоста. Оно объявляет данные; ссылки на сервер или клиент у него нет.
Заменить базовое поведение. Методы спецификации и базовые теги результатов отклоняются при создании (initialize и вовсе зарезервирован за механизмом запуска); привязка уведомления, перекрытая базовым словарём, вместо этого замолкает с предупреждением.
Зарегистрироваться с опозданием. После того как MCPServer(...) или Client(...) вернул управление, набор расширений уже не меняется.

Если вы боретесь с этими стенами, вы пишете не расширение. Вы пишете форк. Стены и есть главное достоинство: тот, кто читает extensions=[Apps(), Stamps()], знает всё, чего эти два расширения могли коснуться.

Расширения

Использование расширения

Написание собственного расширения

Идентификатор

Добавление инструментов

Обслуживание собственных методов

Клиентская сторона

Перехват tools/call

Использование клиентского расширения

Написание клиентского расширения

Глаголы расширения

Чего расширение не может

Перехват `tools/call`