Add a client extension API (#3034)

docs_src/extensions/tutorial007.py

@@ -0,0 +1,50 @@
+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