Extensions
Maschinelle Übersetzung
Diese Seite wurde automatisch aus der englischen Dokumentation übersetzt, und die englische Seite ist die maßgebliche Fassung. Wenn sich etwas falsch liest, erklärt Übersetzungen, wie du es melden kannst.
Eine Extension ist ein optionales Bündel von MCP-Verhalten hinter einem einzigen Identifier.
Auf einem Server kann sie Tools, Ressourcen und neue Request-Methoden beisteuern, und sie kann tools/call umhüllen. Auf einem Client kann sie zusätzliche Ergebnisformen von tools/call für sich beanspruchen und Vendor-Benachrichtigungen beobachten. Jede Seite kündigt sie unter ihrem eigenen capabilities.extensions an, und für alle, die nicht darum gebeten haben, ändert sich nichts. Das ist der Vertrag (SEP-2133), und er hat eine goldene Regel: Extensions sind standardmäßig aus.
Eine Extension verwenden
Übergib Instanzen bei der Konstruktion:
from mcp.server.apps import Apps
from mcp.server.mcpserver import MCPServer
mcp = MCPServer("demo", extensions=[Apps()])
Fertig. Der Server kündigt jetzt io.modelcontextprotocol/ui unter capabilities.extensions an und bedient alles, was die Extension beisteuert.
Apps ist die eingebaute Referenz-Extension und bekommt eine eigene Seite: MCP Apps.
Note
Extensions stehen bei der Konstruktion fest. Es gibt kein add_extension, das du später aufrufen könntest: Die Capability-Map eines Servers sollte sich nicht ändern, während Clients mit ihm verbunden sind.
Die Capability-Map reist mit server/discover, und das ist ein Pfad von 2026-07-28. Ein Legacy-initialize-Handshake hat keinen Platz dafür, also sieht ein Legacy-Client die Extension schlicht nicht. Plane das ein: Eine Extension ergänzt einen Server, sie darf nicht der einzige Weg sein, auf dem der Server nutzbar ist.
Eine eigene Extension schreiben
Leite von Extension ab und überschreibe nur, was du brauchst. Jede Methode hat eine Standardimplementierung.
Der Identifier
from mcp.server.extension import Extension
class Stamps(Extension):
identifier = "com.example/stamps"
Der Identifier ist ein vendor-prefix/name-String nach der _meta-Schlüsselgrammatik der Spezifikation: durch Punkte getrennte Labels (jedes beginnt mit einem Buchstaben und endet mit einem Buchstaben oder einer Ziffer), ein Schrägstrich, dann der Name. Er wird bei der Definition der Klasse validiert, ein Tippfehler wartet also nicht darauf, dass ein Server startet:
TypeError: Stamps.identifier must be a `vendor-prefix/name` string
(reverse-DNS prefix required), got 'stamps'
Nimm als Präfix eine Domain, die du kontrollierst. io.modelcontextprotocol/* ist Extensions vorbehalten, die das MCP-Projekt selbst spezifiziert.
Tools beisteuern
Die kleinste nützliche Extension besteht aus einem Tool und einer Settings-Map:
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()gibtToolBindings zurück. Der Server registriert jedes einzelne genau so, als hättest du selbstmcp.add_tool(...)aufgerufen: dieselbe Schema-Generierung, dieselbeContext-Injection, alles gleich.settings()ist der Wert, der untercapabilities.extensions["com.example/stamps"]angekündigt wird. Gib{}zurück (den Standardwert), um die Extension ohne Settings anzukündigen.- Die Extension bekommt den Server nie in die Hand. Sie deklariert ihre Beiträge als Daten;
MCPSerververarbeitet sie. Es gibt keinself.server, das sie verändern könnte.
Und main() ist der Beweis, ein In-Memory-Client direkt gegen mcp:
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')]
Eigene Methoden bedienen
Eine Extension kann neue Request-Methoden registrieren: eigene Verben, bedient neben denen der Spezifikation:
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']
SearchParamsleitet vonRequestParamsab, sodass der_meta-Umschlag von 2026 einheitlich geparst wird und dein Handler validierte Parameter bekommt, nie ein rohes Dict. Begrenze, was der Client kontrolliert:Field(ge=1, le=100)weist ein absurdeslimitzurück, bevor dein Code irgendetwas dafür alloziert.require_client_extension(ctx, EXTENSION_ID)ist die Schranke: Ein Client, der die Extension nicht deklariert hat, bekommt den Fehler-32021(missing required client capability), samt der maschinenlesbarenrequiredCapabilities-Payload, die die Spezifikation verlangt.protocol_versions=frozenset({"2026-07-28"})heftet die Methode an genau eine Protokollversion auf der Leitung. Bei jeder anderen Version bekommt der ClientMETHOD_NOT_FOUND, genau so, als gäbe es die Methode dort nicht. Für diesen Client gibt es sie auch nicht.
Methoden sind strikt additiv. Das SDK erzwingt das bei der Konstruktion, nicht zur Laufzeit:
- Ein
MethodBindingfür eine in der Spezifikation definierte Methode (tools/list,completion/complete, ...) löstValueErroraus, wenn das Binding konstruiert wird. Kernverben gehören dem Server. - Zwei Extensions, die dieselbe Methode binden, lösen eine Exception aus, sobald sich die zweite registriert. Last-write-wins ist genau der Weg, auf dem Plugins einander beschädigen; das machen wir nicht.
- Ein leeres
protocol_versions-Set löst ebenfalls eine Exception aus: Eine Methode, die nie bedient werden kann, ist ein Bug, keine Konfiguration.
Die Client-Seite
Das main() derselben Datei ist die ganze Client-Geschichte, beide Hälften davon:
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)])deklariert die Extension. Die Deklarationen werden zuClientCapabilities.extensions: Auf einer 2026-07-28-Verbindung reist die Map im_meta-Umschlag jedes einzelnen Requests, der Server sieht sie also bei jedem Request; auf einer Legacy-Verbindung reist sie mit deminitialize-Handshake. Dem Server-Code ist das egal:require_client_extension(ctx, ...)undctx.session.check_client_capability(...)lesen auf beiden Pfaden die richtige Quelle.- Vendor-Methoden steigen eine Schicht tiefer zu
client.session.send_request(...)hinab;Clientbekommt nur für Verben der Spezifikation eigene Methoden.send_requestakzeptiert jedeRequest-Unterklasse, der Vendor-Request geht also unverändert durch.
tools/call abfangen
Der eine eingreifende Hook. Überschreibe intercept_tool_call, um einen Tool-Aufruf zu beobachten, kurzzuschließen oder zu verhindern:
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
paramssind die validiertenCallToolRequestParams: Du bekommstparams.nameundparams.arguments, ohne rohes JSON anzufassen. Sie entscheiden auch, welcher Tool-Aufruf läuft: Reichst du übercall_nexteinen umgeschriebenen Kontext weiter, ändert das, was der Handler aufctxsieht, nicht den Tool-Aufruf selbst. Das Umschreiben von Requests auf Leitungsebene gehört in die Middleware.call_next(ctx)führt den Rest der Kette aus und gibt das Ergebnis des Handlers zurück. Gib es unverändert zurück (beobachten), gib etwas anderes zurück (ersetzen) oder löse einenMCPErroraus (ablehnen). Was immer du zurückgibst, wird wie jedes Handler-Ergebnis serialisiert, einschließlich desserverInfo-Identitätsstempels der 2026er-Generation, ein kurzschließender Interceptor erzeugt also nie eine anonyme oder vom Schema abweichende Response.- Bei mehreren Extensions schachteln sich die Interceptors in Registrierungsreihenfolge: Die erste Extension in
extensions=[...]liegt ganz außen. - Die Standardimplementierung reicht einfach durch, und ein Server, dessen Extensions diesen Hook nie überschreiben, behält den nackten
tools/call-Handler unangetastet. Du zahlst nicht für das, was du nicht nutzt.
Der Hook umhüllt tools/call und sonst nichts. Für alles, was jede Nachricht betrifft, nimm Middleware. Dafür ist sie da.
Eine Client-Extension verwenden
Eine Client-Extension ist derselbe Vertrag von der konsumierenden Seite: ein Bündel clientseitigen Verhaltens hinter einem einzigen Identifier. Übergib Instanzen an Client(extensions=[...]) und rufe Tools ganz normal auf:
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", ...) gibt ein gewöhnliches CallToolResult zurück, wie jeder andere Aufruf. Was die Extension geändert hat: Der Server darf buy jetzt mit einer receipt-Ergebnisform statt mit einem endgültigen Ergebnis beantworten, und Receipts bringt sie zu Ende (hier, indem sie den Beleg mit einem Folgeaufruf einlöst), bevor call_tool zurückkehrt. An der Aufrufstelle bewegt sich nichts.
Lass die Extension weg, und nichts davon existiert: Die Schranke des Servers weist einen Client ab, der sie nicht deklariert hat (Fehler -32021), und eine beanspruchte Form von einem Server, der die Schranke überspringt, fällt durch die Validierung, genau wie die Spezifikation es für einen unbekannten resultType verlangt. Standardmäßig aus, an beiden Enden der Leitung.
Um einen Identifier ohne clientseitiges Verhalten anzukündigen (der Server prüft die Capability, der Client tut nichts, wie beim Search-Client oben), nimm advertise():
from mcp.client import advertise
client = Client(mcp, extensions=[advertise("com.example/search")])
Eine Client-Extension schreiben
Leite von ClientExtension ab und überschreibe nur, was du brauchst. Drei Arten von Beiträgen, jede mit einer Standardimplementierung: settings(), claims() und notifications().
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')]
- Der Identifier folgt derselben Grammatik wie auf dem Server und wird validiert, wenn die Klasse definiert wird.
claims()gibtResultClaims zurück: ein Tag auf der Leitung, das Model, das es parst, und der Resolver, der es zu Ende bringt. Das Model muss das Tag mitresult_type: Literal["receipt"]festlegen und darf nicht von den Kern-Ergebnistypen des Verbs ableiten; beides wird erzwungen, wenn der Claim konstruiert wird. Vendor-Felder wiereceipt_tokengehen unverändert über die Leitung: Eine ersetzte Form erreicht den Client wortwörtlich.- Der Resolver erhält das geparste Model und einen
ClaimContext;ctx.sessionist derselbe öffentliche Griff wieclient.session, Folgeaufrufe sind also gewöhnliche Session-Aufrufe. Er gibt das normaleCallToolResultdes Verbs zurück. settings()ist der Wert, der unterClientCapabilities.extensions[identifier]angekündigt wird, einmal bei der Konstruktion vonClientgelesen.
notifications() deklariert Vendor-Benachrichtigungen des Servers, die beobachtet werden sollen:
def notifications(self) -> Sequence[NotificationBinding[Any]]:
return [NotificationBinding(method="notifications/receipts", params_type=ReceiptEvent, handler=self.on_receipt)]
Der Handler erhält validierte Parameter, eine Benachrichtigung nach der anderen, in Dispatch-Reihenfolge. Er beobachtet; ein Veto einlegen oder antworten kann er nicht.
Zwei stille Regeln. Claims sind nur auf 2026-07-28-Verbindungen aktiv, und die Capability-Ankündigung folgt ihnen: Auf einer Legacy-Verbindung lösen sich die Claims auf, und der Identifier fällt mit ihnen aus der Ankündigung heraus, der Client kündigt also nie eine Extension an, deren Formen er zurückweisen würde. Und wenn du die beanspruchte Form selbst statt des Resolvers haben willst, rufe client.session.call_tool(..., allow_claimed=True) auf; ohne dieses Flag löst eine beanspruchte Form, die bei einem Aufrufer auf Session-Ebene ankommt, UnexpectedClaimedResult aus.
Extension-Verben
Die eigenen Request-Methoden einer Extension brauchen keine clientseitige Registrierung. Ein Vendor-Request-Typ leitet von mcp.types.Request ab und geht durch client.session.send_request, wie in Eigene Methoden bedienen. Eine Ergänzung: Wenn ein Params-Schlüssel im Mcp-Name-Header mitreisen muss (Extension-Spezifikationen wie Tasks verlangen das für ihre Verben), deklariert der Request-Typ name_param:
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
Die Session spiegelt params["jobId"] auf jedem Sendepfad in Mcp-Name, und ein fehlender Wert scheitert laut, statt einen erforderlichen Header stillschweigend wegzulassen.
Was eine Extension nicht kann
Die Fläche für Beiträge ist absichtlich geschlossen. Auf dem Server: Settings, Tools, Ressourcen, Methoden, ein tools/call-Interceptor. Auf dem Client: Settings, Result-Claims, Notification-Bindings. Eine Extension kann nicht:
- In den Host hineingreifen. Sie deklariert Daten; sie hält keine Referenz auf Server oder Client.
- Kernverhalten ersetzen. Methoden der Spezifikation und Kern-Ergebnistags werden bei der Konstruktion abgewiesen (
initializereserviert der Runner von vornherein für sich); ein Notification-Binding, das vom Kernvokabular überdeckt wird, verstummt stattdessen mit einer Warnung. - Sich nachträglich registrieren. Sobald
MCPServer(...)oderClient(...)zurückgekehrt ist, ist die Menge der Extensions, wie sie ist.
Wenn du gegen diese Wände ankämpfst, schreibst du keine Extension. Du schreibst einen Fork. Die Wände sind das Feature: Wer extensions=[Apps(), Stamps()] liest, weiß alles, was diese beiden angefasst haben können.