docs: stop presenting the in-memory client as the way to connect (#3443)

This commit is contained in:
Max 2026-09-05 04:15:58 +01:00 • committed by GitHub
commit 7bb486a10f
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
8 changed files with 803 additions and 598 deletions

View file

@ -1,7 +1,6 @@
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
@ -24,12 +23,3 @@ class Stamps(Extension):
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')]

View file

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

View file

@ -1,11 +1,9 @@
from collections.abc import Sequence
from typing import Any, Literal
from typing import Any
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
@ -22,11 +20,6 @@ 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)])
@ -49,11 +42,3 @@ class Search(Extension):
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']

View file

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

View file

@ -1,9 +1,6 @@
from collections.abc import Sequence
from typing import Any, Literal
from typing import Any
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
@ -11,13 +8,6 @@ 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."""
@ -35,18 +25,6 @@ class ReceiptIssuer(Extension):
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()])
@ -60,10 +38,3 @@ def buy(item: str) -> types.CallToolResult:
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')]

View file

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

View file

@ -1,9 +1,7 @@
from collections.abc import Sequence
from typing import Any, Literal
from typing import Any
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
@ -19,12 +17,6 @@ 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")
@ -39,11 +31,3 @@ class Jobs(Extension):
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

View file

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