wehub-resource-sync
49b9bb6724
Deploy Docs / deploy-docs (push) Failing after 1s
Conformance Tests / client-conformance (push) Failing after 3s
Conformance Tests / server-conformance (push) Failing after 1s
GitHub Actions Security Analysis / zizmor (push) Failing after 1s
CI / checks (push) Failing after 59m20s
CI / all-green (push) Waiting to run
111 lines
4.6 KiB
Python
111 lines
4.6 KiB
Python
"""serve_one / serve_connection mechanics: the kernel drivers a transport entry composes.
|
|
|
|
`handle_one()` is the modern single-exchange recipe (`Connection.from_envelope`
|
|
+ `serve_one` → raw result dict). `main()` is the loop recipe
|
|
(`JSONRPCDispatcher` + `Connection.for_loop` + `serve_connection`) — what
|
|
`Server.run()` does for stdio. Both drivers take a `lowlevel.Server`, so this is
|
|
a lowlevel-only story: `MCPServer` has no public accessor for its underlying
|
|
`Server` yet.
|
|
"""
|
|
|
|
from collections.abc import Mapping
|
|
from dataclasses import dataclass, field
|
|
from typing import Any
|
|
|
|
import anyio
|
|
import mcp_types as types
|
|
from mcp_types.version import LATEST_MODERN_VERSION
|
|
|
|
from mcp.server.connection import Connection # deep-path import; shorter re-export planned
|
|
from mcp.server.context import ServerRequestContext
|
|
from mcp.server.lowlevel import Server
|
|
from mcp.server.runner import serve_connection, serve_one # deep-path import; shorter re-export planned
|
|
from mcp.server.stdio import stdio_server
|
|
from mcp.shared.exceptions import NoBackChannelError
|
|
from mcp.shared.jsonrpc_dispatcher import JSONRPCDispatcher
|
|
from mcp.shared.transport_context import TransportContext
|
|
|
|
__all__ = ["SingleExchangeContext", "build_server", "handle_one"]
|
|
|
|
|
|
def build_server() -> Server[Any]:
|
|
async def list_tools(
|
|
ctx: ServerRequestContext[Any], params: types.PaginatedRequestParams | None
|
|
) -> types.ListToolsResult:
|
|
return types.ListToolsResult(
|
|
tools=[types.Tool(name="add", description="Add two integers.", input_schema={"type": "object"})]
|
|
)
|
|
|
|
async def call_tool(ctx: ServerRequestContext[Any], params: types.CallToolRequestParams) -> types.CallToolResult:
|
|
assert params.name == "add" and params.arguments is not None
|
|
total = params.arguments["a"] + params.arguments["b"]
|
|
return types.CallToolResult(content=[types.TextContent(text=str(total))], structured_content={"result": total})
|
|
|
|
return Server("serve-one-example", on_list_tools=list_tools, on_call_tool=call_tool)
|
|
|
|
|
|
@dataclass
|
|
class SingleExchangeContext:
|
|
"""Minimal `DispatchContext` for one inbound request with no back-channel.
|
|
|
|
A custom transport entry hand-builds one of these per request. The SDK
|
|
ships no public concrete class for this yet; this is the structural minimum.
|
|
"""
|
|
|
|
request_id: int | str | None
|
|
transport: TransportContext = field(default_factory=lambda: TransportContext(kind="custom", can_send_request=False))
|
|
message_metadata: None = None
|
|
can_send_request: bool = False
|
|
cancel_requested: anyio.Event = field(default_factory=anyio.Event)
|
|
|
|
async def send_raw_request(self, method: str, params: Mapping[str, Any] | None, opts: Any = None) -> dict[str, Any]:
|
|
raise NoBackChannelError(method)
|
|
|
|
async def notify(self, method: str, params: Mapping[str, Any] | None, opts: Any = None) -> None:
|
|
return None
|
|
|
|
async def progress(self, progress: float, total: float | None = None, message: str | None = None) -> None:
|
|
return None
|
|
|
|
|
|
async def handle_one(
|
|
server: Server[Any], method: str, params: Mapping[str, Any], *, lifespan_state: Any
|
|
) -> dict[str, Any]:
|
|
"""Serve exactly one modern-era request and return its raw result dict.
|
|
|
|
Reads the envelope from `params._meta` (the 2026 wire shape), builds a
|
|
born-ready `Connection.from_envelope`, and drives `serve_one`. The transport
|
|
entry enters `server.lifespan(server)` once and threads `lifespan_state` to
|
|
every call — never enter the lifespan per-request.
|
|
"""
|
|
meta = params.get("_meta", {})
|
|
connection = Connection.from_envelope(
|
|
meta.get(types.PROTOCOL_VERSION_META_KEY, LATEST_MODERN_VERSION),
|
|
meta.get(types.CLIENT_INFO_META_KEY),
|
|
meta.get(types.CLIENT_CAPABILITIES_META_KEY),
|
|
)
|
|
return await serve_one(
|
|
server,
|
|
SingleExchangeContext(request_id=1),
|
|
method,
|
|
params,
|
|
connection=connection,
|
|
lifespan_state=lifespan_state,
|
|
)
|
|
|
|
|
|
async def main() -> None:
|
|
"""Serve over stdio by building the dispatcher + Connection by hand (loop mode)."""
|
|
server = build_server()
|
|
async with server.lifespan(server) as lifespan_state:
|
|
async with stdio_server() as (read_stream, write_stream):
|
|
dispatcher: JSONRPCDispatcher[TransportContext] = JSONRPCDispatcher(
|
|
read_stream, write_stream, inline_methods=frozenset({"initialize"})
|
|
)
|
|
connection = Connection.for_loop(dispatcher)
|
|
await serve_connection(server, dispatcher, connection=connection, lifespan_state=lifespan_state)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
anyio.run(main)
|