conformance: classify MCP spans before GenAI compatibility matching

tools/gen-ai/runner/src/genai_conformance/_coverage.py

@@ -1,7 +1,7 @@
 # Copyright The OpenTelemetry Authors
 # SPDX-License-Identifier: Apache-2.0
 
-"""How to recognise a GenAI span type.
+"""How to recognise a GenAI or MCP span type.
 
 A run is reduced by asking, per registry span type, which of that type's
 attributes were present. The registry declares a type's attributes but not how
@@ -14,6 +14,11 @@
 
 from typing import Any, Callable, Mapping
 
+_MCP_SPAN_TYPES = {
+    "client": "mcp.client",
+    "server": "mcp.server",
+}
+
 # Which operation names belong to which span type.
 _OPERATION_NAMES = {
     "gen_ai.create_agent.client": {"create_agent"},
@@ -66,7 +71,9 @@ def classify_span(
     ) -> set[str]:
         """The span types a span belongs to.
 
-        ``gen_ai.operation.name`` names the type when it is set. A span that
+        ``mcp.method.name`` identifies an MCP span before GenAI compatibility
+        attributes are considered. ``gen_ai.operation.name`` names the type
+        otherwise. A span that
         omits it is recognised by the attributes only its type carries — but a
         span that names its operation *is* that operation, whatever else it
         carries, so an inference span holding ``gen_ai.agent.id`` is not an
@@ -76,10 +83,20 @@ def classify_span(
         signature.
         """
         del span_name
-        operation = str(attributes.get("gen_ai.operation.name", "")).lower()
         present = {
             name for name, value in attributes.items() if value is not None
         }
+        spans = coverage_model["spans"]
+
+        mcp_span_type = _MCP_SPAN_TYPES.get(span_kind.lower())
+        if (
+            "mcp.method.name" in present
+            and mcp_span_type is not None
+            and mcp_span_type in spans
+        ):
+            return {mcp_span_type}
+
+        operation = str(attributes.get("gen_ai.operation.name", "")).lower()
 
         named = {
             span_type
@@ -94,7 +111,6 @@ def classify_span(
 
         # Span kind separates otherwise identical types, e.g. an agent invoked
         # over the wire (client) from one running in-process (internal).
-        spans = coverage_model["spans"]
         of_this_kind = {
             span_type
             for span_type in matched

tools/gen-ai/runner/src/genai_conformance/policies/genai_span_validation.rego

@@ -100,6 +100,7 @@ deny contains _span_finding(
 	]),
 ) if {
 	input.sample.span
+	not _is_mcp_span(input.sample.span)
 	op := _attr_value(input.sample.span, "gen_ai.operation.name")
 	keyed_attr := _span_name_keyed_attr[op]
 	not _valid_op_and_attr_span_name(input.sample.span, op, keyed_attr)
@@ -167,6 +168,7 @@ deny contains _span_finding(
 	),
 ) if {
 	input.sample.span
+	not _is_mcp_span(input.sample.span)
 	op := _attr_value(input.sample.span, "gen_ai.operation.name")
 	expected := _expected_for_op(op, input.sample.span.kind)
 	some attr_name in expected
@@ -215,6 +217,7 @@ deny contains _span_finding(
 	),
 ) if {
 	input.sample.span
+	not _is_mcp_span(input.sample.span)
 	op := _attr_value(input.sample.span, "gen_ai.operation.name")
 	expected_kinds := _expected_kinds_for_op[op]
 	not expected_kinds[input.sample.span.kind]
@@ -254,6 +257,17 @@ deny contains _span_finding(
 
 # Span attributes arrive as `[{"name": ..., "value": ..., "type": ...}]`.
 
+# MCP spans can carry GenAI compatibility attributes for tool calls, but their
+# shape and name are governed by the MCP span conventions.
+_mcp_span_type["client"] := "mcp.client"
+_mcp_span_type["server"] := "mcp.server"
+
+_is_mcp_span(span) if {
+	_has_attr(span, "mcp.method.name")
+	span_type := _mcp_span_type[span.kind]
+	data["coverage-model"].spans[span_type]
+}
+
 # True when the span has an attribute named `name`.
 _has_attr(span, name) if {
 	some attr in span.attributes

tools/gen-ai/runner/tests/test_genai_coverage.py

@@ -34,15 +34,28 @@ def _classify_span(model):
     return DOMAIN.classifier(model)
 
 
+@pytest.fixture(name="mcp_classify_span", scope="module")
+def _mcp_classify_span():
+    """Classify without requiring the externally resolved registry model."""
+    return DOMAIN.classifier(
+        {
+            "spans": {
+                "gen_ai.execute_tool.internal": {"kind": "internal"},
+                "mcp.client": {"kind": "client"},
+                "mcp.server": {"kind": "server"},
+            }
+        }
+    )
+
+
 @pytest.fixture(name="reduce_for")
 def _reduce_for(model):
     """Reduce a run that emitted the given signals carrying every attribute."""
 
     def build(span_types=(), events=(), metrics=(), entities=()):
         def signals(names, section):
             return {
-                name: set(model[section][name]["attributes"])
-                for name in names
+                name: set(model[section][name]["attributes"]) for name in names
             }
 
         resources = {
@@ -71,6 +84,57 @@ def test_the_operation_name_names_the_span_type(classify_span) -> None:
     ) == {"gen_ai.inference.client"}
 
 
+@pytest.mark.parametrize(
+    ("span_kind", "span_type"),
+    [("client", "mcp.client"), ("server", "mcp.server")],
+)
+def test_mcp_method_takes_precedence_over_genai_compatibility_attributes(
+    mcp_classify_span, span_kind, span_type
+) -> None:
+    assert mcp_classify_span(
+        "tools/call get_weather",
+        span_kind,
+        {
+            "mcp.method.name": "tools/call",
+            "gen_ai.operation.name": "execute_tool",
+            "gen_ai.tool.name": "get_weather",
+        },
+    ) == {span_type}
+
+
+def test_mcp_method_does_not_reclassify_an_internal_tool_span(
+    mcp_classify_span,
+) -> None:
+    assert mcp_classify_span(
+        "execute_tool get_weather",
+        "internal",
+        {
+            "mcp.method.name": "tools/call",
+            "gen_ai.operation.name": "execute_tool",
+            "gen_ai.tool.name": "get_weather",
+        },
+    ) == {"gen_ai.execute_tool.internal"}
+
+
+def test_mcp_method_falls_back_when_registry_has_no_mcp_span_type() -> None:
+    classify_span = DOMAIN.classifier(
+        {
+            "spans": {
+                "gen_ai.execute_tool.internal": {"kind": "internal"},
+            }
+        }
+    )
+
+    assert classify_span(
+        "tools/call get_weather",
+        "client",
+        {
+            "mcp.method.name": "tools/call",
+            "gen_ai.operation.name": "execute_tool",
+        },
+    ) == {"gen_ai.execute_tool.internal"}
+
+
 def test_a_span_without_an_operation_name_is_identified_by_its_attributes(
     classify_span,
 ) -> None:
@@ -154,7 +218,9 @@ def test_every_declared_event_is_recordable(model, reduce_for) -> None:
 
 def test_every_declared_entity_is_recordable(model, reduce_for) -> None:
     declared = [
-        name for name in model.get("entities", {}) if name.startswith("gen_ai.")
+        name
+        for name in model.get("entities", {})
+        if name.startswith("gen_ai.")
     ]
 
     recorded = reduce_for(entities=declared)["entities"]

tools/gen-ai/runner/tests/test_genai_policy.py

@@ -0,0 +1,162 @@
+# Copyright The OpenTelemetry Authors
+# SPDX-License-Identifier: Apache-2.0
+
+"""GenAI span advice policy behavior around MCP compatibility attributes."""
+
+from __future__ import annotations
+
+import copy
+import json
+import subprocess
+from pathlib import Path
+from typing import Any
+
+import pytest
+
+from genai_conformance import DOMAIN
+from opentelemetry.conformance import WeaverNotInstalledError, check_weaver
+
+_POLICY = (
+    Path(__file__).parents[1]
+    / "src/genai_conformance/policies/genai_span_validation.rego"
+)
+_SHAPE_POLICY_IDS = {
+    "genai_expected_attribute_missing",
+    "genai_span_kind_unexpected",
+    "genai_span_name_format",
+}
+
+
+def _span(name: str, kind: str) -> dict[str, Any]:
+    return {
+        "span": {
+            "name": name,
+            "kind": kind,
+            "attributes": [
+                {"name": "mcp.method.name", "value": "tools/call"},
+                {"name": "gen_ai.operation.name", "value": "execute_tool"},
+                {"name": "gen_ai.tool.name", "value": "get_weather"},
+            ],
+        }
+    }
+
+
+def _run_policy(
+    root: Path,
+    registry: Path,
+    model: dict[str, Any],
+    samples: list[dict[str, Any]],
+) -> dict[tuple[str, str], set[str]]:
+    input_path = root / "input.json"
+    model_path = root / "coverage-model.json"
+    report_dir = root / "report"
+    input_path.write_text(json.dumps(samples), encoding="utf-8")
+    model_path.write_text(json.dumps(model), encoding="utf-8")
+
+    result = subprocess.run(
+        [
+            "weaver",
+            "registry",
+            "live-check",
+            "--quiet",
+            "--registry",
+            str(registry),
+            "--input-source",
+            str(input_path),
+            "--advice-policies",
+            str(_POLICY),
+            "--advice-data",
+            str(model_path),
+            "--format",
+            "json",
+            "--fail-on",
+            "none",
+            "--no-stream",
+            "--output",
+            str(report_dir),
+        ],
+        capture_output=True,
+        text=True,
+        check=False,
+    )
+    assert result.returncode == 0, result.stderr
+
+    report: dict[str, Any] = json.loads(
+        (report_dir / "live_check.json").read_text(encoding="utf-8")
+    )
+    return {
+        (span["kind"], span["name"]): {
+            advice["id"]
+            for advice in span["live_check_result"]["all_advice"]
+            if advice["id"] in _SHAPE_POLICY_IDS
+        }
+        for sample in report["samples"]
+        if (span := sample.get("span")) is not None
+    }
+
+
+@pytest.fixture(scope="module")
+def policy_advice(
+    tmp_path_factory: pytest.TempPathFactory,
+) -> tuple[dict[tuple[str, str], set[str]], dict[tuple[str, str], set[str]]]:
+    try:
+        check_weaver()
+        registry = DOMAIN.registry
+        model = copy.deepcopy(DOMAIN.coverage_model)
+    except (WeaverNotInstalledError, OSError, RuntimeError) as error:
+        pytest.skip(f"GenAI policy test unavailable: {error}")
+
+    with_mcp = _run_policy(
+        tmp_path_factory.mktemp("genai-policy-mcp"),
+        registry,
+        model,
+        [
+            _span("tools/call get_weather", "client"),
+            _span("tools/call get_weather", "server"),
+            _span("wrong get_weather", "internal"),
+        ],
+    )
+
+    model["spans"].pop("mcp.client", None)
+    model["spans"].pop("mcp.server", None)
+    without_mcp = _run_policy(
+        tmp_path_factory.mktemp("genai-policy-fallback"),
+        registry,
+        model,
+        [_span("tools/call get_weather", "client")],
+    )
+    return with_mcp, without_mcp
+
+
+def test_mcp_client_and_server_spans_skip_genai_shape_rules(
+    policy_advice: tuple[
+        dict[tuple[str, str], set[str]], dict[tuple[str, str], set[str]]
+    ],
+) -> None:
+    with_mcp, _ = policy_advice
+
+    assert with_mcp[("client", "tools/call get_weather")] == set()
+    assert with_mcp[("server", "tools/call get_weather")] == set()
+
+
+def test_internal_span_with_mcp_attribute_keeps_genai_shape_rules(
+    policy_advice: tuple[
+        dict[tuple[str, str], set[str]], dict[tuple[str, str], set[str]]
+    ],
+) -> None:
+    with_mcp, _ = policy_advice
+
+    assert (
+        "genai_span_name_format" in with_mcp[("internal", "wrong get_weather")]
+    )
+
+
+def test_missing_mcp_definitions_keep_genai_shape_rules(
+    policy_advice: tuple[
+        dict[tuple[str, str], set[str]], dict[tuple[str, str], set[str]]
+    ],
+) -> None:
+    _, without_mcp = policy_advice
+    advice = without_mcp[("client", "tools/call get_weather")]
+
+    assert advice == {"genai_span_name_format"}