This is an automated email from the ASF dual-hosted git repository.
FreeOnePlus pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris-mcp-server.git
The following commit(s) were added to refs/heads/master by this push:
new abbb3f5 feat: propagate sanitized MCP trace context (#149)
abbb3f5 is described below
commit abbb3f54c76b3f1d5ee1145c1203a7adb31e622c
Author: Yijia Su <[email protected]>
AuthorDate: Thu Jul 30 13:33:50 2026 +0800
feat: propagate sanitized MCP trace context (#149)
---
CHANGELOG.md | 3 +
README.md | 25 +++
doris_mcp_server/protocol.py | 5 +
doris_mcp_server/trace_context.py | 248 ++++++++++++++++++++++
doris_mcp_server/utils/redaction.py | 5 +
test/protocol/test_trace_context.py | 374 ++++++++++++++++++++++++++++++++++
test/protocol/trace_context_server.py | 156 ++++++++++++++
7 files changed, 816 insertions(+)
diff --git a/CHANGELOG.md b/CHANGELOG.md
index a4fa10f..b8d6721 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -39,6 +39,9 @@ under **Unreleased** until a new version is selected and
published.
Streamable HTTP and stdio.
- HMAC-authenticated explicit state handles with principal, scope, resource,
expiry, and shared-worker key binding instead of protocol-session state.
+- W3C `traceparent`, `tracestate`, and `baggage` propagation from request
+ `_meta`, with value-safe validation, credential-like baggage redaction,
+ per-request isolation, and no trace metadata in model-facing results.
- Real Doris process tests covering Streamable HTTP and stdio.
### Changed
diff --git a/README.md b/README.md
index 307a24a..006da86 100644
--- a/README.md
+++ b/README.md
@@ -535,6 +535,31 @@ Stdio carries the same JSON-RPC request metadata in the
message body, but it
does not use HTTP headers. Do not write logs or other diagnostics to stdout in
stdio mode; stdout is reserved for MCP protocol messages.
+### OpenTelemetry Trace Context
+
+Clients may propagate the W3C `traceparent`, `tracestate`, and `baggage`
+carrier fields in `params._meta`, as defined by
+[MCP SEP-414](https://modelcontextprotocol.io/seps/414-request-meta),
+[W3C Trace Context](https://www.w3.org/TR/trace-context/), and
+[W3C Baggage](https://www.w3.org/TR/baggage/). The same message-level carrier
+works on Streamable HTTP and stdio; these values are not separate MCP HTTP
+headers.
+
+When an OpenTelemetry provider and exporter are configured, each MCP operation
+span is parented to the valid incoming trace context. The active context is
+available to instrumented downstream work for the lifetime of that operation
+and is reset before the next request. Trace carrier fields are never passed to
+the Doris tool, resource, or prompt managers and are never copied into model
+content or structured results.
+
+The server validates trace carrier values before the SDK propagator sees them.
+Malformed, oversized, duplicate, or orphaned fields are ignored independently,
+and warning logs identify only the field name—never its supplied value.
+Values under credential-like baggage keys such as `token`, `secret`, or
+`authorization` are replaced with `[REDACTED]` before propagation. `baggage`
+can still contain other deployment-sensitive correlation data, so clients
+should send only values approved by their telemetry data-handling policy.
+
### List Pagination
`resources/list`, `tools/list`, and `prompts/list` return at most
diff --git a/doris_mcp_server/protocol.py b/doris_mcp_server/protocol.py
index 1972fec..423655d 100644
--- a/doris_mcp_server/protocol.py
+++ b/doris_mcp_server/protocol.py
@@ -73,6 +73,7 @@ from .state_handles import (
DEFAULT_STATE_HANDLE_TTL_SECONDS,
StateHandleCodec,
)
+from .trace_context import TraceContextSanitizingMiddleware
from .utils.redaction import (
redact_error_payload,
redact_sensitive_text,
@@ -558,6 +559,10 @@ def create_doris_mcp_server(
on_list_prompts=list_prompts,
on_get_prompt=get_prompt,
)
+ # The SDK's OpenTelemetry middleware extracts W3C carrier values. Run the
+ # value-safe sanitizer before it so malformed tracestate/baggage cannot be
+ # echoed by dependency warning logs and never reaches the handler layer.
+ server.middleware.insert(0, TraceContextSanitizingMiddleware(logger))
async def hide_unhandled_errors(
ctx: ServerRequestContext,
diff --git a/doris_mcp_server/trace_context.py
b/doris_mcp_server/trace_context.py
new file mode 100644
index 0000000..489ada1
--- /dev/null
+++ b/doris_mcp_server/trace_context.py
@@ -0,0 +1,248 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""Safe W3C trace-context handling for MCP request ``_meta``."""
+
+from __future__ import annotations
+
+import logging
+import re
+from collections.abc import Mapping
+from dataclasses import replace
+from typing import Any, cast
+from urllib.parse import unquote_plus
+
+from mcp.server import ServerRequestContext
+from mcp.server.context import CallNext, HandlerResult
+from mcp.types import RequestParamsMeta
+from opentelemetry.trace import get_current_span
+from opentelemetry.trace.propagation.tracecontext import (
+ TraceContextTextMapPropagator,
+)
+
+from .utils.redaction import REDACTED, is_sensitive_key
+
+TRACEPARENT_META_KEY = "traceparent"
+TRACESTATE_META_KEY = "tracestate"
+BAGGAGE_META_KEY = "baggage"
+
+_TRACEPARENT_MAX_LENGTH = 512
+_TRACESTATE_MAX_LENGTH = 512
+_TRACESTATE_MAX_MEMBERS = 32
+_BAGGAGE_MAX_LENGTH = 8192
+_BAGGAGE_MAX_MEMBER_LENGTH = 4096
+_BAGGAGE_MAX_MEMBERS = 180
+
+# W3C Trace Context section 3.3.1.3. The expression intentionally mirrors
+# the specification's ASCII bounds rather than accepting arbitrary Unicode.
+_TRACESTATE_MEMBER_RE = re.compile(
+ r"(?P<key>"
+ r"[a-z][_0-9a-z\-*/]{0,255}"
+ r"|[a-z0-9][_0-9a-z\-*/]{0,240}@[a-z][_0-9a-z\-*/]{0,13}"
+ r")="
+ r"[\x20-\x2b\x2d-\x3c\x3e-\x7e]{0,255}"
+ r"[\x21-\x2b\x2d-\x3c\x3e-\x7e]"
+ r"[ \t]*"
+)
+_LIST_DELIMITER_RE = re.compile(r"[ \t]*,[ \t]*")
+
+# W3C Baggage section 3.2.1. The value expression includes optional
+# semicolon-delimited properties and remains aligned with the OTel W3C
+# propagator's accepted wire surface.
+_BAGGAGE_KEY_RE = re.compile(
+ r"[\x21\x23-\x27\x2a\x2b\x2d\x2e"
+ r"\x30-\x39\x41-\x5a\x5e-\x7a\x7c\x7e]+"
+)
+_BAGGAGE_VALUE_RE = re.compile(
+ r"[\x21\x23-\x2b\x2d-\x3a\x3c-\x5b\x5d-\x7e]*"
+)
+_BAGGAGE_PROPERTY_RE = re.compile(
+ r"[ \t]*"
+ r"[\x21\x23-\x27\x2a\x2b\x2d\x2e"
+ r"\x30-\x39\x41-\x5a\x5e-\x7a\x7c\x7e]+"
+ r"(?:[ \t]*=[ \t]*"
+ r"[\x21\x23-\x2b\x2d-\x3a\x3c-\x5b\x5d-\x7e]*)?"
+ r"[ \t]*"
+)
+_TRACEPARENT_PROPAGATOR = TraceContextTextMapPropagator()
+
+
+def _valid_traceparent(value: object) -> bool:
+ if (
+ not isinstance(value, str)
+ or not value
+ or len(value) > _TRACEPARENT_MAX_LENGTH
+ ):
+ return False
+ context = _TRACEPARENT_PROPAGATOR.extract(
+ {TRACEPARENT_META_KEY: value}
+ )
+ return get_current_span(context).get_span_context().is_valid
+
+
+def _valid_tracestate(value: object) -> bool:
+ if (
+ not isinstance(value, str)
+ or not value
+ or len(value) > _TRACESTATE_MAX_LENGTH
+ ):
+ return False
+ members = _LIST_DELIMITER_RE.split(value)
+ if len(members) > _TRACESTATE_MAX_MEMBERS:
+ return False
+ keys: set[str] = set()
+ for member in members:
+ match = _TRACESTATE_MEMBER_RE.fullmatch(member)
+ if match is None or match.group("key") in keys:
+ return False
+ keys.add(match.group("key"))
+ return True
+
+
+def _sanitize_baggage(value: object) -> str | None:
+ if (
+ not isinstance(value, str)
+ or not value
+ or len(value) > _BAGGAGE_MAX_LENGTH
+ ):
+ return None
+ members = _LIST_DELIMITER_RE.split(value)
+ if len(members) > _BAGGAGE_MAX_MEMBERS:
+ return None
+ sanitized_members: list[str] = []
+ decoded_keys: set[str] = set()
+ for member in members:
+ if not member or len(member) > _BAGGAGE_MAX_MEMBER_LENGTH:
+ return None
+ try:
+ key, raw_value = member.split("=", 1)
+ except ValueError:
+ return None
+ if _BAGGAGE_KEY_RE.fullmatch(key) is None:
+ return None
+ decoded_key = unquote_plus(key).strip()
+ if decoded_key in decoded_keys:
+ return None
+ decoded_keys.add(decoded_key)
+ value_and_properties = raw_value.split(";")
+ if _BAGGAGE_VALUE_RE.fullmatch(value_and_properties[0]) is None:
+ return None
+ if any(
+ _BAGGAGE_PROPERTY_RE.fullmatch(item) is None
+ for item in value_and_properties[1:]
+ ):
+ return None
+ if is_sensitive_key(decoded_key):
+ sanitized_members.append(f"{key}={REDACTED}")
+ continue
+
+ sanitized_properties: list[str] = []
+ for item in value_and_properties[1:]:
+ if "=" not in item:
+ sanitized_properties.append(item)
+ continue
+ property_key, _property_value = item.split("=", 1)
+ if is_sensitive_key(unquote_plus(property_key).strip()):
+ sanitized_properties.append(
+ f"{property_key.rstrip()}={REDACTED}"
+ )
+ else:
+ sanitized_properties.append(item)
+ sanitized_members.append(
+ ";".join(
+ [
+ f"{key}={value_and_properties[0]}",
+ *sanitized_properties,
+ ]
+ )
+ )
+ return ",".join(sanitized_members)
+
+
+def sanitize_trace_meta(
+ meta: Mapping[str, Any] | None,
+ *,
+ logger: logging.Logger,
+) -> RequestParamsMeta | None:
+ """Drop malformed trace fields without logging their untrusted values."""
+ if meta is None:
+ return None
+
+ sanitized = dict(meta)
+ traceparent_valid = (
+ TRACEPARENT_META_KEY not in sanitized
+ or _valid_traceparent(sanitized[TRACEPARENT_META_KEY])
+ )
+ if not traceparent_valid:
+ sanitized.pop(TRACEPARENT_META_KEY, None)
+ logger.warning(
+ "Ignoring invalid MCP trace metadata field %s",
+ TRACEPARENT_META_KEY,
+ )
+
+ if TRACESTATE_META_KEY in sanitized:
+ tracestate_valid = (
+ traceparent_valid
+ and TRACEPARENT_META_KEY in sanitized
+ and _valid_tracestate(sanitized[TRACESTATE_META_KEY])
+ )
+ if not tracestate_valid:
+ sanitized.pop(TRACESTATE_META_KEY, None)
+ logger.warning(
+ "Ignoring invalid MCP trace metadata field %s",
+ TRACESTATE_META_KEY,
+ )
+
+ if BAGGAGE_META_KEY in sanitized:
+ sanitized_baggage = _sanitize_baggage(sanitized[BAGGAGE_META_KEY])
+ if sanitized_baggage is None:
+ sanitized.pop(BAGGAGE_META_KEY, None)
+ logger.warning(
+ "Ignoring invalid MCP trace metadata field %s",
+ BAGGAGE_META_KEY,
+ )
+ else:
+ sanitized[BAGGAGE_META_KEY] = sanitized_baggage
+
+ return cast(RequestParamsMeta, sanitized)
+
+
+class TraceContextSanitizingMiddleware:
+ """Sanitize trace carrier fields before the SDK OTel middleware runs."""
+
+ def __init__(self, logger: logging.Logger) -> None:
+ self._logger = logger
+
+ async def __call__(
+ self,
+ ctx: ServerRequestContext[Any, Any],
+ call_next: CallNext,
+ ) -> HandlerResult:
+ sanitized_meta = sanitize_trace_meta(ctx.meta, logger=self._logger)
+ if sanitized_meta == ctx.meta:
+ return await call_next(ctx)
+
+ sanitized_params = ctx.params
+ if ctx.params is not None and "_meta" in ctx.params:
+ sanitized_params = dict(ctx.params)
+ sanitized_params["_meta"] = sanitized_meta or {}
+ return await call_next(
+ replace(
+ ctx,
+ meta=sanitized_meta,
+ params=sanitized_params,
+ )
+ )
diff --git a/doris_mcp_server/utils/redaction.py
b/doris_mcp_server/utils/redaction.py
index ce8abe1..c3007b9 100644
--- a/doris_mcp_server/utils/redaction.py
+++ b/doris_mcp_server/utils/redaction.py
@@ -122,6 +122,11 @@ def _normalized_key(key: object) -> str:
return re.sub(r"[^a-z0-9]", "", str(key).casefold())
+def is_sensitive_key(key: object) -> bool:
+ """Return whether a semantic field name is treated as credential data."""
+ return _normalized_key(key) in _SENSITIVE_KEYS
+
+
def redact_sql_literals(value: str) -> str:
"""Remove values and comments from SQL while retaining diagnostic shape."""
if not _SQL_KEYWORD_RE.search(value):
diff --git a/test/protocol/test_trace_context.py
b/test/protocol/test_trace_context.py
new file mode 100644
index 0000000..10c2677
--- /dev/null
+++ b/test/protocol/test_trace_context.py
@@ -0,0 +1,374 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""OpenTelemetry request ``_meta`` propagation and redaction tests."""
+
+from __future__ import annotations
+
+import json
+import logging
+import secrets
+import sys
+from collections.abc import Generator
+from contextlib import contextmanager
+from pathlib import Path
+from typing import Any
+
+import httpx2
+import mcp.shared._otel as mcp_otel
+import pytest
+from mcp import Client, StdioServerParameters
+from mcp.client.stdio import stdio_client
+from opentelemetry.baggage import get_all
+from opentelemetry.trace import (
+ NonRecordingSpan,
+ SpanContext,
+ TraceFlags,
+ get_current_span,
+ use_span,
+)
+
+from doris_mcp_server.protocol import create_transport_security
+from doris_mcp_server.trace_context import sanitize_trace_meta
+from test.protocol.test_mcp_v2_protocol import (
+ create_test_server,
+ modern_tool_headers,
+ modern_tool_request,
+)
+
+_TRACE_ID = "0af7651916cd43dd8448eb211c80319c"
+_PARENT_SPAN_ID = "00f067aa0ba902b7"
+
+
+class _CapturingTracer:
+ def __init__(self) -> None:
+ self.records: list[dict[str, Any]] = []
+
+ @contextmanager
+ def start_as_current_span(
+ self,
+ name: str,
+ *,
+ context: Any = None,
+ attributes: dict[str, Any] | None = None,
+ **kwargs: Any,
+ ) -> Generator[NonRecordingSpan]:
+ del attributes, kwargs
+ parent = get_current_span(context).get_span_context()
+ trace_id = parent.trace_id if parent.is_valid else
secrets.randbits(128) or 1
+ child_context = SpanContext(
+ trace_id=trace_id,
+ span_id=secrets.randbits(64) or 1,
+ is_remote=False,
+ trace_flags=parent.trace_flags if parent.is_valid else
TraceFlags(0),
+ trace_state=parent.trace_state if parent.is_valid else None,
+ )
+ baggage = get_all(context=context)
+ self.records.append(
+ {
+ "name": name,
+ "parentTraceId": (
+ f"{parent.trace_id:032x}" if parent.is_valid else None
+ ),
+ "parentSpanId": (
+ f"{parent.span_id:016x}" if parent.is_valid else None
+ ),
+ "baggageCount": len(baggage),
+ "redactedBaggageCount": sum(
+ value == "[REDACTED]" for value in baggage.values()
+ ),
+ }
+ )
+ span = NonRecordingSpan(child_context)
+ with use_span(span, end_on_exit=False):
+ yield span
+
+
+def _trace_meta(*, baggage: str, tracestate: str = "vendor=opaque") -> dict:
+ return {
+ "traceparent": f"00-{_TRACE_ID}-{_PARENT_SPAN_ID}-01",
+ "tracestate": tracestate,
+ "baggage": baggage,
+ }
+
+
[email protected](
+ ("meta", "remaining_trace_keys"),
+ [
+ (
+ {
+ "traceparent": "not-a-traceparent",
+ "tracestate": "vendor=opaque",
+ "baggage": "tenant=blue",
+ },
+ {"baggage"},
+ ),
+ ({"tracestate": "vendor=opaque"}, set()),
+ (
+ {
+ "traceparent": (
+ f"00-{_TRACE_ID}-{_PARENT_SPAN_ID}-01"
+ ),
+ "tracestate": "vendor=one,vendor=two",
+ },
+ {"traceparent"},
+ ),
+ (
+ {
+ "traceparent": (
+ f"00-{_TRACE_ID}-{_PARENT_SPAN_ID}-01"
+ ),
+ "baggage": "missing-value",
+ },
+ {"traceparent"},
+ ),
+ (
+ {
+ "traceparent": (
+ f"00-{_TRACE_ID}-{_PARENT_SPAN_ID}-01"
+ ),
+ "baggage": "key=" + "x" * 8193,
+ },
+ {"traceparent"},
+ ),
+ ],
+)
+def test_trace_meta_sanitizer_drops_only_invalid_carrier_fields(
+ meta: dict[str, Any],
+ remaining_trace_keys: set[str],
+ caplog: pytest.LogCaptureFixture,
+):
+ secret = "sanitizer-must-not-log-values"
+ supplied = {**meta, "custom": secret}
+ logger = logging.getLogger("test.trace-context-sanitizer")
+ caplog.set_level(logging.WARNING)
+
+ sanitized = sanitize_trace_meta(supplied, logger=logger)
+
+ assert sanitized is not None
+ assert sanitized["custom"] == secret
+ assert (
+ {"traceparent", "tracestate", "baggage"} & set(sanitized)
+ == remaining_trace_keys
+ )
+ assert secret not in caplog.text
+
+
+def test_trace_meta_sanitizer_redacts_sensitive_baggage_values():
+ logger = logging.getLogger("test.trace-context-sanitizer")
+
+ sanitized = sanitize_trace_meta(
+ {
+ "baggage": (
+ "tenant=blue,secret=must-not-propagate,"
+ "token%2Dsecret=also-private,"
+ "region=west;authorization=private;readonly"
+ )
+ },
+ logger=logger,
+ )
+
+ assert sanitized is not None
+ assert sanitized["baggage"] == (
+ "tenant=blue,secret=[REDACTED],token%2Dsecret=[REDACTED],"
+ "region=west;authorization=[REDACTED];readonly"
+ )
+
+
[email protected]
+async def test_http_propagates_trace_context_without_model_or_log_leakage(
+ monkeypatch: pytest.MonkeyPatch,
+ caplog: pytest.LogCaptureFixture,
+):
+ tracer = _CapturingTracer()
+ monkeypatch.setattr(mcp_otel, "_tracer", tracer)
+ app = create_test_server().streamable_http_app(
+ json_response=True,
+ stateless_http=True,
+ host="127.0.0.1",
+ transport_security=create_transport_security("127.0.0.1"),
+ )
+ valid_secret = "http-valid-baggage-secret"
+ invalid_secret = "http-invalid-baggage-secret"
+ caplog.set_level(logging.WARNING)
+
+ async with (
+ app.router.lifespan_context(app),
+ httpx2.ASGITransport(app) as transport,
+ httpx2.AsyncClient(
+ transport=transport,
+ base_url="http://127.0.0.1:3000";,
+ ) as client,
+ ):
+ traced_request = modern_tool_request(1, "echo", {"value": "safe"})
+ traced_request["params"]["_meta"].update(
+ _trace_meta(baggage=f"tenant=blue,secret={valid_secret}")
+ )
+ traced = await client.post(
+ "/mcp",
+ json=traced_request,
+ headers=modern_tool_headers("echo"),
+ )
+ assert traced.status_code == 200
+ assert traced.json()["result"]["structuredContent"] == {
+ "name": "echo",
+ "arguments": {"value": "safe"},
+ }
+ assert _TRACE_ID not in traced.text
+ assert valid_secret not in traced.text
+
+ sanitized_request = modern_tool_request(2, "echo", {"value": "safe"})
+ sanitized_request["params"]["_meta"].update(
+ _trace_meta(
+ baggage=f"secret={invalid_secret},missing-value",
+ tracestate=f"vendor=opaque,{invalid_secret}",
+ )
+ )
+ sanitized = await client.post(
+ "/mcp",
+ json=sanitized_request,
+ headers=modern_tool_headers("echo"),
+ )
+ assert sanitized.status_code == 200
+ assert invalid_secret not in sanitized.text
+
+ untraced = await client.post(
+ "/mcp",
+ json=modern_tool_request(3, "echo", {"value": "safe"}),
+ headers=modern_tool_headers("echo"),
+ )
+ assert untraced.status_code == 200
+
+ tool_records = [
+ record
+ for record in tracer.records
+ if record["name"] == "tools/call echo"
+ ]
+ assert tool_records == [
+ {
+ "name": "tools/call echo",
+ "parentTraceId": _TRACE_ID,
+ "parentSpanId": _PARENT_SPAN_ID,
+ "baggageCount": 2,
+ "redactedBaggageCount": 1,
+ },
+ {
+ "name": "tools/call echo",
+ "parentTraceId": _TRACE_ID,
+ "parentSpanId": _PARENT_SPAN_ID,
+ "baggageCount": 0,
+ "redactedBaggageCount": 0,
+ },
+ {
+ "name": "tools/call echo",
+ "parentTraceId": None,
+ "parentSpanId": None,
+ "baggageCount": 0,
+ "redactedBaggageCount": 0,
+ },
+ ]
+ log_text = caplog.text
+ assert "Ignoring invalid MCP trace metadata field tracestate" in log_text
+ assert "Ignoring invalid MCP trace metadata field baggage" in log_text
+ assert valid_secret not in log_text
+ assert invalid_secret not in log_text
+
+
[email protected]
+async def test_true_subprocess_stdio_propagates_and_sanitizes_trace_meta(
+ tmp_path: Path,
+):
+ server_script = Path(__file__).with_name("trace_context_server.py")
+ observations = tmp_path / "trace-observations.jsonl"
+ log_path = tmp_path / "trace-context.log"
+ server_params = StdioServerParameters(
+ command=sys.executable,
+ args=[str(server_script)],
+ env={
+ "DORIS_MCP_TRACE_OBSERVATIONS": str(observations),
+ "DORIS_MCP_TRACE_LOG": str(log_path),
+ },
+ )
+ valid_secret = "stdio-valid-baggage-secret"
+ invalid_secret = "stdio-invalid-baggage-secret"
+
+ async with Client(stdio_client(server_params)) as client:
+ traced = await client.call_tool(
+ "echo",
+ {},
+ meta=_trace_meta(
+ baggage=f"tenant=blue,secret={valid_secret}"
+ ),
+ )
+ assert traced.structured_content == {"ok": True}
+ traced_wire = json.dumps(
+ traced.model_dump(by_alias=True, mode="json"),
+ sort_keys=True,
+ )
+ assert _TRACE_ID not in traced_wire
+ assert valid_secret not in traced_wire
+
+ sanitized = await client.call_tool(
+ "echo",
+ {},
+ meta=_trace_meta(
+ baggage=f"secret={invalid_secret},missing-value",
+ tracestate=f"vendor=opaque,{invalid_secret}",
+ ),
+ )
+ assert sanitized.structured_content == {"ok": True}
+
+ untraced = await client.call_tool("echo", {})
+ assert untraced.structured_content == {"ok": True}
+
+ records = [
+ json.loads(line)
+ for line in observations.read_text(encoding="utf-8").splitlines()
+ ]
+ tool_records = [
+ record for record in records if record["name"] == "tools/call echo"
+ ]
+ assert tool_records == [
+ {
+ "name": "tools/call echo",
+ "parentTraceId": _TRACE_ID,
+ "parentSpanId": _PARENT_SPAN_ID,
+ "baggageCount": 2,
+ "redactedBaggageCount": 1,
+ },
+ {
+ "name": "tools/call echo",
+ "parentTraceId": _TRACE_ID,
+ "parentSpanId": _PARENT_SPAN_ID,
+ "baggageCount": 0,
+ "redactedBaggageCount": 0,
+ },
+ {
+ "name": "tools/call echo",
+ "parentTraceId": None,
+ "parentSpanId": None,
+ "baggageCount": 0,
+ "redactedBaggageCount": 0,
+ },
+ ]
+ observed_text = observations.read_text(encoding="utf-8")
+ log_text = log_path.read_text(encoding="utf-8")
+ assert valid_secret not in observed_text
+ assert invalid_secret not in observed_text
+ assert "Ignoring invalid MCP trace metadata field tracestate" in log_text
+ assert "Ignoring invalid MCP trace metadata field baggage" in log_text
+ assert valid_secret not in log_text
+ assert invalid_secret not in log_text
diff --git a/test/protocol/trace_context_server.py
b/test/protocol/trace_context_server.py
new file mode 100644
index 0000000..ad5e768
--- /dev/null
+++ b/test/protocol/trace_context_server.py
@@ -0,0 +1,156 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""True-subprocess stdio fixture for MCP OpenTelemetry propagation."""
+
+from __future__ import annotations
+
+import asyncio
+import json
+import logging
+import os
+import secrets
+from collections.abc import Generator
+from contextlib import contextmanager
+from pathlib import Path
+from typing import Any
+
+import mcp.shared._otel as mcp_otel
+from mcp.server.stdio import stdio_server
+from mcp.types import GetPromptResult, Prompt, Resource, Tool
+from opentelemetry.baggage import get_all
+from opentelemetry.trace import (
+ NonRecordingSpan,
+ SpanContext,
+ TraceFlags,
+ get_current_span,
+ use_span,
+)
+
+from doris_mcp_server import __version__
+from doris_mcp_server.protocol import create_doris_mcp_server
+
+
+class _ResourcesManager:
+ async def list_resources(self) -> list[Resource]:
+ return []
+
+ async def read_resource(self, uri: str) -> str:
+ return json.dumps({"uri": uri})
+
+
+class _ToolsManager:
+ async def list_tools(self) -> list[Tool]:
+ return [
+ Tool(
+ name="echo",
+ description="Return a fixed response without request
metadata.",
+ input_schema={"type": "object", "properties": {}},
+ )
+ ]
+
+ async def call_tool(self, name: str, arguments: dict[str, Any]) -> str:
+ del name, arguments
+ return json.dumps({"ok": True})
+
+
+class _PromptsManager:
+ async def list_prompts(self) -> list[Prompt]:
+ return []
+
+ async def get_prompt(
+ self,
+ name: str,
+ arguments: dict[str, Any],
+ ) -> GetPromptResult:
+ del name, arguments
+ raise AssertionError("prompt fixture is not called")
+
+
+class _ObservationTracer:
+ """Small API-only tracer that records propagation without an SDK
exporter."""
+
+ def __init__(self, output_path: Path) -> None:
+ self._output_path = output_path
+
+ @contextmanager
+ def start_as_current_span(
+ self,
+ name: str,
+ *,
+ context: Any = None,
+ attributes: dict[str, Any] | None = None,
+ **kwargs: Any,
+ ) -> Generator[NonRecordingSpan]:
+ del attributes, kwargs
+ parent = get_current_span(context).get_span_context()
+ trace_id = parent.trace_id if parent.is_valid else
secrets.randbits(128) or 1
+ child_context = SpanContext(
+ trace_id=trace_id,
+ span_id=secrets.randbits(64) or 1,
+ is_remote=False,
+ trace_flags=parent.trace_flags if parent.is_valid else
TraceFlags(0),
+ trace_state=parent.trace_state if parent.is_valid else None,
+ )
+ baggage = get_all(context=context)
+ record = {
+ "name": name,
+ "parentTraceId": (
+ f"{parent.trace_id:032x}" if parent.is_valid else None
+ ),
+ "parentSpanId": (
+ f"{parent.span_id:016x}" if parent.is_valid else None
+ ),
+ "baggageCount": len(baggage),
+ "redactedBaggageCount": sum(
+ value == "[REDACTED]" for value in baggage.values()
+ ),
+ }
+ with self._output_path.open("a", encoding="utf-8") as stream:
+ stream.write(json.dumps(record, sort_keys=True) + "\n")
+ span = NonRecordingSpan(child_context)
+ with use_span(span, end_on_exit=False):
+ yield span
+
+
+async def main() -> None:
+ observation_path = Path(os.environ["DORIS_MCP_TRACE_OBSERVATIONS"])
+ log_path = Path(os.environ["DORIS_MCP_TRACE_LOG"])
+ logging.basicConfig(
+ filename=log_path,
+ level=logging.WARNING,
+ force=True,
+ )
+ mcp_otel._tracer = _ObservationTracer(observation_path)
+ logger = logging.getLogger("doris_mcp_server.trace_context_fixture")
+ server = create_doris_mcp_server(
+ resources_manager=_ResourcesManager(),
+ tools_manager=_ToolsManager(),
+ prompts_manager=_PromptsManager(),
+ name="doris-mcp-trace-context-test",
+ version=__version__,
+ logger=logger,
+ )
+ async with stdio_server() as (read_stream, write_stream):
+ await server.run(
+ read_stream,
+ write_stream,
+ server.create_initialization_options(),
+ )
+
+
+if __name__ == "__main__":
+ asyncio.run(main())
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
