Розширення
Машинний переклад
Цю сторінку перекладено автоматично з англомовної документації, і основною версією є англомовна сторінка. Якщо щось читається неправильно, на сторінці Переклади пояснено, як про це повідомити.
Розширення — це набір поведінки MCP, який вмикається лише на явний запит і стоїть за одним ідентифікатором.
На сервері воно може додавати інструменти, ресурси й нові методи запитів, а також обгортати tools/call. На клієнті — заявляти додаткові форми результату tools/call і спостерігати за вендорськими сповіщеннями. Кожна сторона оголошує розширення у власному capabilities.extensions, і для тих, хто про це не просив, нічого не змінюється. Такий контракт (SEP-2133), і в нього одне золоте правило: розширення за замовчуванням вимкнені.
Використання розширення
Передайте екземпляри під час створення:
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.
Додавання інструментів
Найменше корисне розширення — один інструмент і карта налаштувань:
from collections.abc import Sequence
from typing import Any
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()])
tools()повертає об'єктиToolBinding. Сервер реєструє кожен із них точно так, ніби ви самі викликалиmcp.add_tool(...): те саме генерування схеми, те саме впровадженняContext, усе те саме.settings()— це значення, що оголошується вcapabilities.extensions["com.example/stamps"]. Поверніть{}(типове значення), щоб оголосити розширення без налаштувань.- Розширення ніколи не отримує сервер. Воно оголошує свій внесок як дані;
MCPServerїх споживає. Ніякогоself.server, який можна було б змінювати, немає.
Запустіть його через HTTP, а доказом буде клієнт:
uv run mcp run server.py --transport streamable-http
import anyio
from mcp import Client
async def main() -> None:
async with Client("http://localhost:8000/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(type='text', text='[stamped] hello', annotations=None, meta=None)]
if __name__ == "__main__":
anyio.run(main)
Кожен server.py на цій сторінці запускається цією командою, а кожен client.py працює поруч із ним: python client.py у другому терміналі.
Обслуговування власних методів
Розширення може реєструвати нові методи запитів: власні дієслова, які обслуговуються поруч із методами специфікації:
from collections.abc import Sequence
from typing import Any
from pydantic import Field
import mcp.types as types
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]
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()])
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теж викидає виняток: метод, який ніколи не може бути обслужений, — це помилка, а не конфігурація.
Клієнтська сторона
Клієнт — окрема програма, і в ній обидві половини клієнтської частини:
from typing import Literal
import anyio
import mcp.types as types
from mcp import Client
from mcp.client import advertise
EXTENSION_ID = "com.example/search"
class SearchParams(types.RequestParams):
query: str
limit: int = 10
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 main() -> None:
async with Client("http://localhost:8000/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']
if __name__ == "__main__":
anyio.run(main)
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, тож вендорський запит проходить як є. SearchRequestі дві моделі, які він несе, — це мережевий контракт розширення, тож клієнт оголошує їх для себе сам. Опубліковане розширення постачало б їх у пакеті, який імпортують обидві сторони.
Перехоплення tools/call
Єдиний хук-перехоплювач. Перевизначте intercept_tool_call, щоб спостерігати за викликом інструмента, завершувати його достроково або забороняти:
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. Саме для цього воно й існує.
Використання клієнтського розширення
Клієнтське розширення — той самий контракт з боку споживача: набір клієнтської поведінки за одним ідентифікатором. Сервер тут відповідає на buy не товаром, а квитанцією, яку треба погасити, — і лише клієнту, що оголосив розширення:
from typing import Any
import mcp.types as types
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 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"}
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}"
На клієнті передайте екземпляри в Client(extensions=[...]) і викликайте інструменти як зазвичай:
from collections.abc import Sequence
from typing import Any, Literal
import anyio
import mcp.types as types
from mcp import Client
from mcp.client import ClaimContext, ClientExtension, ResultClaim
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 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})
async def main() -> None:
async with Client("http://localhost:8000/mcp", extensions=[Receipts()]) as client:
result = await client.call_tool("buy", {"item": "lamp"})
print(result.content)
# [TextContent(type='text', text='goods for r-117', annotations=None, meta=None)]
if __name__ == "__main__":
anyio.run(main)
call_tool("buy", ...) повертає звичайний CallToolResult, як і будь-який інший виклик. Що змінило розширення: тепер сервер може відповісти на buy формою результату receipt замість остаточного результату, а Receipts доводить її до кінця (тут — погашаючи квитанцію наступним викликом), перш ніж call_tool поверне значення. У місці виклику нічого не змінюється.
Приберіть розширення — і нічого з цього не існує: шлагбаум сервера відмовляє клієнту, який його не оголосив (помилка -32021), а заявлена форма від сервера, що обходить шлагбаум, не проходить перевірку — точно так, як специфікація вимагає для нерозпізнаного resultType. Вимкнено за замовчуванням, з обох кінців з'єднання.
Щоб оголосити ідентифікатор без жодної клієнтської поведінки (сервер перевіряє можливість, клієнт нічого не робить, як у клієнті пошуку вище), використовуйте advertise():
from mcp.client import advertise
client = Client("http://localhost:8000/mcp", extensions=[advertise("com.example/search")])
Написання клієнтського розширення
Успадкуйте ClientExtension і перевизначте лише те, що потрібно. Три види внеску, кожен із типовою реалізацією: settings(), claims() і notifications().
from collections.abc import Sequence
from typing import Any, Literal
import anyio
import mcp.types as types
from mcp import Client
from mcp.client import ClaimContext, ClientExtension, ResultClaim
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 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})
async def main() -> None:
async with Client("http://localhost:8000/mcp", extensions=[Receipts()]) as client:
result = await client.call_tool("buy", {"item": "lamp"})
print(result.content)
# [TextContent(type='text', text='goods for r-117', annotations=None, meta=None)]
if __name__ == "__main__":
anyio.run(main)
- Ідентифікатор підпорядковується тій самій граматиці, що й на сервері, і перевіряється під час визначення класу.
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, як у розділі Обслуговування власних методів. Візьмімо сервер, розширення якого обслуговує одне дієслово про іменоване завдання:
from collections.abc import Sequence
from typing import Any
import mcp.types as types
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
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()])
Одне доповнення на клієнті: коли ключ параметрів мусить передаватися в заголовку Mcp-Name (специфікації розширень, як-от tasks, вимагають цього для своїх дієслів), тип запиту оголошує name_param:
from typing import Literal
import anyio
import mcp.types as types
from mcp import Client
from mcp.client import advertise
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 main() -> None:
async with Client("http://localhost:8000/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
if __name__ == "__main__":
anyio.run(main)
Сесія дзеркалить params["jobId"] у Mcp-Name на кожному шляху надсилання, а відсутнє значення дає гучну помилку замість того, щоб мовчки пропустити обов'язковий заголовок.
Чого розширення не може
Поверхня внеску закрита навмисно. На сервері: налаштування, інструменти, ресурси, методи, один перехоплювач tools/call. На клієнті: налаштування, заявки на результати, прив'язки сповіщень. Розширення не може:
- Лізти в хост. Воно оголошує дані; посилання на сервер чи клієнт у нього немає.
- Замінювати базову поведінку. Методи специфікації та базові теги результатів відхиляються під час створення (
initializeцілком зарезервовано за виконавцем); прив'язка сповіщення, перекрита базовим словником, натомість замовкає з попередженням. - Реєструватися із запізненням. Після того як
MCPServer(...)чиClient(...)повернув керування, набір розширень уже такий, який є.
Якщо ви воюєте з цими стінами, ви пишете не розширення. Ви пишете форк. Стіни — це й є головна перевага: користувач, що читає extensions=[Apps(), Stamps()], знає все, чого ці двоє могли торкнутися.