refactor(mcp): classify traced tool args from a rules file

.gitignore

@@ -13,6 +13,7 @@ docs/*.zip
 
 # Generated MCP server docs (regenerate via `poe mcp-docs-md`)
 docs/mcp-generated/
+/arg_trace_plan.json
 
 # Misc
 .DS_Store

AGENTS.md

@@ -34,3 +34,8 @@ unredacted Cloud message without guidance; this does not add it to LLM Obs metad
 Non-Cloud MCP trace messages are the first line of the exception class's docstring
 (trace-only, exported to Datadog as `error.message`), so keep those first lines fixed text
 with no IDs, values or URLs.
+
+## MCP Argument Tracing
+
+Trace classes come from `airbyte/mcp/arg_trace_rules.yaml`; do not put `TraceArg` markers in tool signatures.
+When `tests/unit_tests/test_mcp_arg_trace_parity.py` reports an unclassified argument, add or adjust a YAML rule; `poe mcp-arg-trace-plan` writes the full plan and each decision source to gitignored `arg_trace_plan.json`.

airbyte/mcp/_arg_trace_rules.py

@@ -0,0 +1,11 @@
+# Copyright (c) 2026 Airbyte, Inc., all rights reserved.
+"""Load MCP argument-tracing rules from the adjacent YAML file."""
+
+from pathlib import Path
+
+import yaml
+
+
+ARG_TRACE_RULES: dict[str, str] = yaml.safe_load(
+    Path(__file__).with_name("arg_trace_rules.yaml").read_text(encoding="utf-8")
+)

airbyte/mcp/_datadog.py

@@ -33,6 +33,7 @@
 )
 from opentelemetry.instrumentation.requests import RequestsInstrumentor
 
+from airbyte.mcp._arg_trace_rules import ARG_TRACE_RULES
 from airbyte.mcp._error_handling import (
     MCP_TOOL_USER_FACING_ERRORS,
     classify_mcp_tool_error,
@@ -88,7 +89,12 @@
 _TOOL_ANNOTATIONS: dict[str, dict[str, Any]] = {}
 _ARG_PREFIX = "airbyte.mcp"
 # Same classification and validation as the OTel backend; intent is exported separately.
-_ARG_TRACER = ArgTracer(_ARG_PREFIX, key=_arg_key, skip=(INTENT_ARG,))
+_ARG_TRACER = ArgTracer(
+    _ARG_PREFIX,
+    rules=ARG_TRACE_RULES,
+    key=_arg_key,
+    skip=(INTENT_ARG,),
+)
 _TOOL_ERROR_MESSAGE = "tool resulted in an error"
 _STATUS_CODE = "airbyte.mcp.upstream.status_code"
 _UPSTREAM_METHOD = "airbyte.mcp.upstream.method"

airbyte/mcp/_otel.py

@@ -21,6 +21,7 @@
 from opentelemetry.trace import SpanKind, Status
 
 from airbyte.constants import CLOUD_API_ROOT, CLOUD_CONFIG_API_ROOT
+from airbyte.mcp._arg_trace_rules import ARG_TRACE_RULES
 from airbyte.mcp._error_handling import classify_mcp_tool_error, mcp_tool_error_reason
 
 
@@ -322,6 +323,7 @@ def install(app: FastMCP, *, environ: Mapping[str, str] | None = None) -> None:
                 error_reason=mcp_tool_error_reason,
                 other_spans=_http_client_span_attributes,
                 arg_key=_arg_key,
+                arg_rules=ARG_TRACE_RULES,
                 exporter=_exporter(backend, environ),
             ),
         ),

airbyte/mcp/_trace_plan.py

@@ -0,0 +1,23 @@
+# Copyright (c) 2026 Airbyte, Inc., all rights reserved.
+"""Write the MCP argument trace plan and its classification sources."""
+
+import asyncio
+import json
+from pathlib import Path
+
+from fastmcp_extensions.otel.middleware import trace_plan
+
+from airbyte.mcp import _otel as observability
+from airbyte.mcp.server import app
+
+
+def main() -> None:
+    """Write the trusted-execution MCP trace plan to the repository root."""
+    observability.install(app, environ={"AIRBYTE_MCP_TRACING_BACKEND": "otel"})
+    plan = asyncio.run(trace_plan(app, include_sources=True))
+    target = Path(__file__).resolve().parents[2] / "arg_trace_plan.json"
+    target.write_text(json.dumps(plan, indent=2, sort_keys=True) + "\n", encoding="utf-8")
+
+
+if __name__ == "__main__":
+    main()

airbyte/mcp/arg_trace_rules.yaml

@@ -0,0 +1,64 @@
+# How each MCP tool argument is recorded in trace spans.
+# Keys are argument names or fnmatch globs; `tool:arg` keys apply to one tool.
+# The first matching key wins. `bool`, `Literal`, and `Enum` arguments are
+# recorded as their value, and dict or model arguments as presence, so they
+# need no rule unless they should be recorded differently.
+
+# Tool-scoped exceptions.
+get_stream_previews:streams: fingerprint
+sync_source_to_cache:streams: fingerprint
+
+# Not recorded.
+workspace_id: omit
+intent: omit
+
+# Identifiers, numbers, and opaque inputs: keyed hash.
+"*_id": hash
+streams: hash
+cursor: hash
+sql: hash
+sql_query: hash
+user_email: hash
+api_args: hash
+manifest_yaml_file: hash
+manifest_yaml_text: hash
+wait_timeout: hash
+min_similarity: hash
+"max_*": hash
+"num_*": hash
+"*_hours": hash
+"*_seconds": hash
+"*_number": hash
+"*_offset": hash
+
+# Free text and names: keyed hash plus a similarity fingerprint.
+name: fingerprint
+"*_name": fingerprint
+"*_names": fingerprint
+name_contains: fingerprint
+selected_streams: fingerprint
+"*_fields": fingerprint
+scenarios: fingerprint
+config_file: fingerprint
+manifest_path: fingerprint
+"*prefix": fingerprint
+"namespace*": fingerprint
+reuse_namespace: fingerprint
+cron_expression: fingerprint
+docker_image: fingerprint
+keyword_filter: fingerprint
+prompt: fingerprint
+query: fingerprint
+search: fingerprint
+section: fingerprint
+
+# Small fixed vocabularies or bounded numbers: raw value.
+limit: value
+page_size: value
+connector_type: value
+entity_type: value
+support_level: value
+min_support_level: value
+sql_dialect: value
+backfill_start_time: value
+defer_credentials: value  # Overrides the `credential` secret-name heuristic.