Extensions
An extension is an opt-in bundle of MCP behaviour behind one identifier.
It can contribute tools, resources, and new request methods, and it can wrap tools/call.
The server advertises it under capabilities.extensions, the client opts in the same way,
and nothing changes for anyone who didn't ask for it. That is the contract (SEP-2133), and
it has one golden rule: extensions are off by default.
Using an extension
Pass instances at construction:
from mcp.server.apps import Apps
from mcp.server.mcpserver import MCPServer
mcp = MCPServer("demo", extensions=[Apps()])
Done. The server now advertises io.modelcontextprotocol/ui under
capabilities.extensions and serves everything the extension contributes.
Apps is the built-in reference extension, and it gets its own page: MCP Apps.
Note
Extensions are fixed at construction. There is no add_extension to call later:
a server's capability map should not change while clients are connected to it.
The capability map rides server/discover, which is a 2026-07-28 path. A legacy
initialize handshake has nowhere to put it, so a legacy client simply doesn't see
the extension. Design for that: an extension augments a server, it must not be the
only way the server is usable.
Writing your own
Subclass Extension and override only what you need. Every method has a default.
The identifier
from mcp.server.extension import Extension
class Stamps(Extension):
identifier = "com.example/stamps"
The identifier is a vendor-prefix/name string following the spec's _meta key
grammar: dot-separated labels (each starts with a letter, ends with a letter or
digit), a slash, then the name. It is validated when the class is defined, so a
typo doesn't wait for a server to boot:
TypeError: Stamps.identifier must be a `vendor-prefix/name` string
(reverse-DNS prefix required), got 'stamps'
Use a domain you control as the prefix. io.modelcontextprotocol/* is for extensions
specified by the MCP project itself.
Contributing tools
The smallest useful extension is one tool and a 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()returnsToolBindings. The server registers each one exactly as if you had calledmcp.add_tool(...)yourself: same schema generation, sameContextinjection, same everything.settings()is the value advertised atcapabilities.extensions["com.example/stamps"]. Return{}(the default) to advertise the extension with no settings.- The extension never receives the server. It declares contributions as data;
MCPServerconsumes them. There is noself.serverto mutate.
And main() is the proof, an in-memory client straight against 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')]
Serving your own methods
An extension can register new request methods: its own verbs, served next to the spec's:
from collections.abc import Sequence
from typing import Any, Literal, cast
import mcp_types as types
from pydantic import Field
from mcp import Client
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={EXTENSION_ID: {}}) as client:
request = SearchRequest(params=SearchParams(query="mcp", limit=3))
result = await client.session.send_request(cast("types.ClientRequest", request), SearchResult)
print(result.items)
# ['mcp-0', 'mcp-1', 'mcp-2']
SearchParamssubclassesRequestParams, so the 2026_metaenvelope parses uniformly and your handler gets validated params, never a raw dict. Bound what the client controls:Field(ge=1, le=100)rejects an absurdlimitbefore your code allocates anything for it.require_client_extension(ctx, EXTENSION_ID)is the gate: a client that did not declare the extension gets the-32021(missing required client capability) error, with the machine-readablerequiredCapabilitiespayload the spec asks for.protocol_versions=frozenset({"2026-07-28"})pins the method to one wire version. At any other version the client getsMETHOD_NOT_FOUND, exactly as if the method didn't exist there. For that client, it doesn't.
Methods are strictly additive. The SDK enforces this at construction, not at runtime:
- A
MethodBindingfor a spec-defined method (tools/list,completion/complete, ...) raisesValueErrorwhen the binding is constructed. Core verbs belong to the server. - Two extensions binding the same method raise when the second one registers. Last-write-wins is how plugins corrupt each other; we don't do that.
- An empty
protocol_versionsset raises too: a method that can never be served is a bug, not a configuration.
The client side
The same file's main() is the whole client story, both halves of it:
from collections.abc import Sequence
from typing import Any, Literal, cast
import mcp_types as types
from pydantic import Field
from mcp import Client
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={EXTENSION_ID: {}}) as client:
request = SearchRequest(params=SearchParams(query="mcp", limit=3))
result = await client.session.send_request(cast("types.ClientRequest", request), SearchResult)
print(result.items)
# ['mcp-0', 'mcp-1', 'mcp-2']
Client(..., extensions={EXTENSION_ID: {}})declares the extension. That map becomesClientCapabilities.extensions: on a 2026-07-28 connection it travels in the per-request_metaenvelope, so the server sees it on every request; on a legacy connection it rides theinitializehandshake. Server code doesn't care which:require_client_extension(ctx, ...)andctx.session.check_client_capability(...)read the right source on both paths.- Vendor methods drop one layer to
client.session.send_request(...);Clientonly grows first-class methods for spec verbs. Thecastis there becausesend_requestis typed against the spec's closed request union.
Intercepting tools/call
The one interceptive hook. Override intercept_tool_call to observe, short-circuit,
or veto a tool call:
import logging
from typing import Any
from mcp_types import CallToolRequestParams
from mcp.server.context import CallNext, HandlerResult, ServerRequestContext
from mcp.server.extension import Extension
from mcp.server.mcpserver import MCPServer
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
paramsis the validatedCallToolRequestParams: you getparams.nameandparams.argumentswithout touching raw JSON.call_next(ctx)runs the rest of the chain. Return its result unchanged (observe), return something else (replace), or raise anMCPError(refuse).- With several extensions, interceptors nest in registration order: the first
extension in
extensions=[...]is outermost. - The default implementation is a pass-through, and a server whose extensions never override this hook installs no middleware at all. You don't pay for what you don't use.
The hook wraps tools/call and nothing else. For every-message concerns, use
Middleware. That is what it is for.
What an extension cannot do
The contribution surface is closed on purpose: settings, tools, resources,
methods, one tools/call interceptor. An extension cannot:
- Reach into the server. It declares data; it holds no server reference.
- Replace core behaviour. Spec methods are rejected at construction, and
initializeis reserved by the runner outright. - Register late. After
MCPServer(...)returns, the extension set is what it is.
If you are fighting these walls, you are not writing an extension. You are writing
a fork. The walls are the feature: a user reading extensions=[Apps(), Stamps()]
knows everything those two can have touched.