Extensions

मशीनी अनुवाद

यह page अंग्रेज़ी documentation से अपने-आप अनुवादित किया गया है, और अंग्रेज़ी page ही प्रामाणिक version है। अगर कुछ गलत लगे, तो अनुवाद page बताता है कि इसकी सूचना कैसे दें।

extension एक identifier के पीछे रखा गया MCP behaviour का opt-in bundle है।

server पर यह tools, resources और नए request methods जोड़ सकता है, और tools/call को wrap कर सकता है। client पर यह tools/call के अतिरिक्त result shapes claim कर सकता है और vendor notifications observe कर सकता है। हर पक्ष अपने-अपने capabilities.extensions के तहत advertise करता है, और जिसने इसे नहीं माँगा उसके लिए कुछ नहीं बदलता। यही contract है (SEP-2133), और इसका एक सुनहरा नियम है: extensions default रूप से बंद रहते हैं।

extension इस्तेमाल करना

construction के समय instances पास करें:

server.py

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

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

हो गया। अब server capabilities.extensions के तहत io.modelcontextprotocol/ui advertise करता है और extension जो कुछ जोड़ता है वह सब serve करता है।

Apps built-in reference extension है, और इसका अपना अलग page है: MCP Apps।

Note

extensions construction के समय ही तय हो जाते हैं। बाद में call करने के लिए कोई add_extension नहीं है: जब clients server से जुड़े हों, तब उसका capability map बदलना नहीं चाहिए।

capability map server/discover के साथ जाता है, जो 2026-07-28 का रास्ता है। legacy initialize handshake में इसे रखने की कोई जगह नहीं है, इसलिए legacy client को extension दिखता ही नहीं। इसे ध्यान में रखकर design करें: extension server को बढ़ाता है, server को इस्तेमाल करने का यही एकमात्र तरीका नहीं होना चाहिए।

अपना extension लिखना

Extension को subclass करें और सिर्फ़ वही override करें जिसकी ज़रूरत हो। हर method का default है।

Identifier

from mcp.server.extension import Extension


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

identifier एक vendor-prefix/name string है जो spec की _meta key grammar का पालन करती है: dot से अलग किए गए labels (हर label अक्षर से शुरू होता है, अक्षर या अंक पर खत्म होता है), फिर एक slash, फिर name। यह class define होते ही validate होता है, इसलिए typo पकड़ने के लिए server के boot होने का इंतज़ार नहीं करना पड़ता:

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

prefix के रूप में ऐसा domain इस्तेमाल करें जो आपके नियंत्रण में हो। io.modelcontextprotocol/* उन extensions के लिए है जिन्हें खुद MCP project specify करता है।

tools जोड़ना

सबसे छोटा काम का extension एक tool और एक settings map है:

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() ToolBindings लौटाता है। server हर एक को ठीक वैसे ही register करता है जैसे आपने खुद mcp.add_tool(...) call किया हो: वही schema generation, वही Context injection, सब कुछ वही।
settings() वह value है जो capabilities.extensions["com.example/stamps"] पर advertise होती है। बिना settings के extension advertise करने के लिए {} (default) लौटाएँ।
extension को server कभी नहीं मिलता। यह अपने योगदान data के रूप में declare करता है; MCPServer उन्हें consume करता है। mutate करने के लिए कोई self.server नहीं है।

और main() इसका सबूत है, सीधे mcp से जुड़ा एक in-memory client:

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')]

अपने methods serve करना

extension नए request methods register कर सकता है: उसके अपने verbs, जो spec के verbs के साथ-साथ serve होते हैं:

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 को subclass करता है, इसलिए 2026 का _meta envelope एक समान तरीके से parse होता है और आपके handler को validated params मिलते हैं, कच्चा dict कभी नहीं। जो client के नियंत्रण में है उसकी सीमा बाँधें: Field(ge=1, le=100) किसी बेतुके limit को तभी reject कर देता है, इससे पहले कि आपका code उसके लिए कुछ allocate करे।
require_client_extension(ctx, EXTENSION_ID) ही gate है: जिस client ने extension declare नहीं किया उसे -32021 (missing required client capability) error मिलता है, साथ में वह machine-readable requiredCapabilities payload जो spec माँगता है।
protocol_versions=frozenset({"2026-07-28"}) method को एक wire version पर pin कर देता है। किसी भी दूसरे version पर client को METHOD_NOT_FOUND मिलता है, ठीक वैसे जैसे method वहाँ मौजूद ही न हो। उस client के लिए, वह है भी नहीं।

methods सख़्ती से additive हैं। SDK इसे construction के समय लागू करता है, runtime पर नहीं:

spec में define किए गए method (tools/list, completion/complete, ...) के लिए MethodBinding बनाते ही ValueError raise होता है। core verbs server के हैं।
एक ही method को bind करने वाले दो extensions हों, तो दूसरा register होते ही raise होता है। plugins एक-दूसरे को last-write-wins से ही खराब करते हैं; हम ऐसा नहीं करते।
खाली protocol_versions set भी raise करता है: जो method कभी serve ही नहीं हो सकता वह bug है, configuration नहीं।

Client side

उसी file का main() ही client की पूरी कहानी है, उसके दोनों हिस्से:

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)]) extension declare करता है। ये declarations ClientCapabilities.extensions बन जाती हैं: 2026-07-28 connection पर यह map हर request के _meta envelope में जाता है, इसलिए server इसे हर request पर देखता है; legacy connection पर यह initialize handshake के साथ जाता है। server code को फ़र्क नहीं पड़ता कि कौन सा: require_client_extension(ctx, ...) और ctx.session.check_client_capability(...) दोनों रास्तों पर सही स्रोत पढ़ते हैं।
vendor methods एक परत नीचे client.session.send_request(...) पर उतरते हैं; Client सिर्फ़ spec verbs के लिए first-class methods जोड़ता है। send_request कोई भी Request subclass स्वीकार करता है, इसलिए vendor request जैसी है वैसी ही चली जाती है।

`tools/call` को intercept करना

यह इकलौता interceptive hook है। tool call को observe, short-circuit या veto करने के लिए intercept_tool_call override करें:

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 validated CallToolRequestParams है: आपको params.name और params.arguments बिना raw JSON छुए मिलते हैं। यही तय करता है कि कौन सा tool call चलेगा: call_next से rewritten context पास करने से वह बदलता है जो handler ctx पर देखता है, tool invocation नहीं। wire-level request rewriting Middleware का काम है।
call_next(ctx) chain का बाकी हिस्सा चलाता है और handler का result लौटाता है। इसे बिना बदले लौटाएँ (observe), कुछ और लौटाएँ (replace), या MCPError raise करें (refuse)। आप जो भी लौटाते हैं वह किसी भी handler result की तरह serialize होता है, 2026 पीढ़ी के serverInfo identity stamp समेत, इसलिए short-circuit करने वाला interceptor कभी anonymous या off-schema response नहीं बनाता।
कई extensions होने पर interceptors registration के क्रम में nest होते हैं: extensions=[...] में पहला extension सबसे बाहर होता है।
default implementation pass-through है, और जिस server के extensions इस hook को कभी override नहीं करते, उसका bare tools/call handler अनछुआ रहता है। जो आप इस्तेमाल नहीं करते उसकी कीमत नहीं चुकाते।

hook tools/call को wrap करता है, और कुछ नहीं। हर message से जुड़ी बातों के लिए Middleware इस्तेमाल करें। वह इसी के लिए है।

client extension इस्तेमाल करना

client extension वही contract है, इस्तेमाल करने वाले पक्ष से: एक identifier के पीछे client-side behaviour का bundle। instances को Client(extensions=[...]) में पास करें और tools सामान्य तरीके से call करें:

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", ...) हर दूसरे call की तरह सादा CallToolResult लौटाता है। extension ने जो बदला: server अब buy का जवाब final result के बजाय receipt result shape से दे सकता है, और call_tool के लौटने से पहले Receipts उसे पूरा कर देता है (यहाँ follow-up call से receipt redeem करके)। call site में कुछ नहीं हिलता।

extension हटा दें तो इनमें से कुछ भी मौजूद नहीं: server का gate उस client को मना कर देता है जिसने इसे declare नहीं किया (error -32021), और gate छोड़ने वाले server से आया claimed shape validation में fail होता है, ठीक वैसे जैसे spec अनजान resultType के लिए माँगता है। default रूप से बंद, wire के दोनों सिरों पर।

बिना किसी client-side behaviour के identifier advertise करने के लिए (server capability पर gate लगाता है, client कुछ नहीं करता, जैसे ऊपर वाले search client में), advertise() इस्तेमाल करें:

from mcp.client import advertise

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

client extension लिखना

ClientExtension को subclass करें और सिर्फ़ वही override करें जिसकी ज़रूरत हो। योगदान के तीन प्रकार, हर एक का default: 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')]

identifier वही grammar मानता है जो server का, और class define होते ही validate होता है।
claims() ResultClaims लौटाता है: एक wire tag, उसे parse करने वाला model, और उसे पूरा करने वाला resolver। model के लिए result_type: Literal["receipt"] से tag pin करना ज़रूरी है और वह verb के core result types को subclass नहीं कर सकता; दोनों बातें claim बनते समय enforce होती हैं। receipt_token जैसे vendor fields wire पर जैसे हैं वैसे जाते हैं: substituted shape client तक हू-ब-हू पहुँचता है।
resolver को parsed model और एक ClaimContext मिलता है; ctx.session वही public handle है जो client.session, इसलिए follow-ups साधारण session calls हैं। यह verb का सामान्य CallToolResult लौटाता है।
settings() वह value है जो ClientCapabilities.extensions[identifier] पर advertise होती है, और Client बनते समय एक बार पढ़ी जाती है।

notifications() observe करने के लिए vendor server notifications declare करता है:

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

handler को validated params एक-एक करके, dispatch के क्रम में मिलते हैं। यह observe करता है; veto या reply नहीं कर सकता।

दो शांत नियम। claims सिर्फ़ 2026-07-28 connections पर सक्रिय रहते हैं, और capability advertisement उन्हीं के पीछे चलता है: legacy connection पर claims गायब हो जाते हैं और identifier भी उनके साथ advertisement से हट जाता है, इसलिए client कभी ऐसा extension advertise नहीं करता जिसके shapes वह खुद reject कर देता। और जब claimed shape resolver के बजाय आपको खुद चाहिए, तो client.session.call_tool(..., allow_claimed=True) call करें; उस flag के बिना, session-tier caller तक पहुँचने वाला claimed shape UnexpectedClaimedResult raise करता है।

Extension verbs

extension के अपने request methods को client-side registration की ज़रूरत नहीं। vendor request type mcp.types.Request को subclass करता है और client.session.send_request से जाता है, जैसा अपने methods serve करना में है। एक बात और: जब किसी params key का Mcp-Name header में जाना ज़रूरी हो (tasks जैसे extension specs अपने verbs के लिए यह माँगते हैं), तो request type name_param declare करता है:

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

session हर send path पर params["jobId"] को Mcp-Name में mirror करता है, और value न होने पर ज़रूरी header चुपचाप छोड़ने के बजाय साफ़ तौर पर fail होता है।

extension क्या नहीं कर सकता

योगदान की surface जानबूझकर बंद रखी गई है। server पर: settings, tools, resources, methods, एक tools/call interceptor। client पर: settings, result claims, notification bindings। extension ये नहीं कर सकता:

host के अंदर पहुँचना। यह data declare करता है; इसके पास server या client का कोई reference नहीं होता।
core behaviour बदलना। spec methods और core result tags construction के समय reject हो जाते हैं (initialize को runner ने पूरी तरह reserve कर रखा है); core vocabulary से ढकी notification binding इसके बजाय warning के साथ चुप हो जाती है।
देर से register करना। MCPServer(...) या Client(...) के लौटने के बाद extension set जैसा है वैसा ही रहता है।

अगर आप इन दीवारों से लड़ रहे हैं, तो आप extension नहीं लिख रहे। आप fork लिख रहे हैं। ये दीवारें ही feature हैं: extensions=[Apps(), Stamps()] पढ़ने वाला user सब कुछ जानता है जिसे ये दोनों छू सकते थे।

Extensions

extension इस्तेमाल करना

अपना extension लिखना

Identifier

tools जोड़ना

अपने methods serve करना

Client side

tools/call को intercept करना

client extension इस्तेमाल करना

client extension लिखना

Extension verbs

extension क्या नहीं कर सकता

`tools/call` को intercept करना