Розширення

Машинний переклад

Цю сторінку перекладено автоматично з англомовної документації, і основною версією є англомовна сторінка. Якщо щось читається неправильно, на сторінці Переклади пояснено, як про це повідомити.

Розширення — це набір поведінки 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, як у розділі Обслуговування власних методів. Одне доповнення: коли ключ параметрів мусить передаватися в заголовку 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`