diff --git a/src/mcp_server_appwrite/auth.py b/src/mcp_server_appwrite/auth.py index acccacc..71f1baa 100644 --- a/src/mcp_server_appwrite/auth.py +++ b/src/mcp_server_appwrite/auth.py @@ -25,7 +25,6 @@ from jwt import PyJWKClient from mcp.server.auth.provider import AccessToken, TokenVerifier -from . import telemetry from .constants import ( CACHE_TTL_SECONDS, DEFAULT_ENDPOINT, @@ -361,7 +360,6 @@ def _verify_sync(self, token: str) -> AccessToken | None: try: unverified = jwt.decode(token, options={"verify_signature": False}) except jwt.PyJWTError: - telemetry.record_auth(outcome="rejected", reason="malformed") return None issuer = unverified.get("iss") @@ -369,7 +367,6 @@ def _verify_sync(self, token: str) -> AccessToken | None: metadata = authorization_server_metadata_sync() except Exception as exc: _log(f"Rejecting token: authorization server discovery failed ({exc}).") - telemetry.record_auth(outcome="rejected", reason="discovery_failed") return None expected_issuer = metadata["issuer"] @@ -378,20 +375,17 @@ def _verify_sync(self, token: str) -> AccessToken | None: f"Rejecting token: issuer {issuer!r} does not match discovered " f"issuer {expected_issuer!r}." ) - telemetry.record_auth(outcome="rejected", reason="issuer_mismatch") return None project_id = project_id_from_issuer(issuer) if not project_id: _log("Rejecting token: issuer is not an Appwrite OAuth issuer.") - telemetry.record_auth(outcome="rejected", reason="issuer_mismatch") return None if project_id != configured_project_id(): _log( f"Rejecting token: issuer project {project_id!r} is not the served " f"project {configured_project_id()!r}." ) - telemetry.record_auth(outcome="rejected", reason="project_mismatch") return None try: @@ -406,12 +400,10 @@ def _verify_sync(self, token: str) -> AccessToken | None: ) except jwt.PyJWTError as exc: _log(f"Rejecting token: verification failed ({exc}).") - telemetry.record_auth(outcome="rejected", reason="signature") return None expected_resource = canonical_resource() if not self._audience_ok(claims.get("aud"), expected_resource): - telemetry.record_auth(outcome="rejected", reason="audience") return None scope_claim = claims.get("scope") or claims.get("scp") or "" @@ -446,18 +438,9 @@ def _audience_ok(self, aud, expected_resource: str) -> bool: return False async def verify_token(self, token: str) -> AccessToken | None: - start = time.monotonic() access_token = await to_thread.run_sync(self._verify_sync, token) - duration = time.monotonic() - start if access_token is None: - # The specific rejection reason was already counted in _verify_sync; - # here we only attach the duration to the rejected outcome. - telemetry.record_auth(outcome="rejected", duration_s=duration, count=False) return None if access_token.expires_at and access_token.expires_at < int(time.time()): - telemetry.record_auth( - outcome="rejected", reason="expired", duration_s=duration - ) return None - telemetry.record_auth(outcome="success", duration_s=duration) return access_token diff --git a/src/mcp_server_appwrite/constants.py b/src/mcp_server_appwrite/constants.py index ee0f345..92c9109 100644 --- a/src/mcp_server_appwrite/constants.py +++ b/src/mcp_server_appwrite/constants.py @@ -151,3 +151,54 @@ def _resolve_server_version() -> str: # --- telemetry ------------------------------------------------------------ ACTIVE_WINDOW_SECONDS = 300.0 # rolling window for "active users/clients" gauges + +# Known MCP clients (normalized: lowercase, whitespace -> "-"). Client names are +# client-controlled input; anything not matching becomes "other" to bound the +# client_id label cardinality. +KNOWN_MCP_CLIENTS = ( + "5ire", + "amp", + "bolt", + "chatgpt", + "cherry-studio", + "claude", + "claude-ai", + "claude-code", + "claude-desktop", + "cline", + "codex", + "codex-cli", + "continue", + "copilot", + "crush", + "cursor", + "deepchat", + "fast-agent", + "gemini", + "gemini-cli", + "github-copilot", + "goose", + "jetbrains", + "kilo-code", + "kiro", + "langchain", + "librechat", + "lm-studio", + "mcp-inspector", + "mcphub", + "n8n", + "opencode", + "openai", + "raycast", + "roo-cline", + "roo-code", + "tome", + "trae", + "visual-studio-code", + "void", + "vscode", + "warp", + "windsurf", + "witsy", + "zed", +) diff --git a/src/mcp_server_appwrite/docs_search.py b/src/mcp_server_appwrite/docs_search.py index d90353e..1b2735c 100644 --- a/src/mcp_server_appwrite/docs_search.py +++ b/src/mcp_server_appwrite/docs_search.py @@ -21,7 +21,6 @@ import mcp.types as types -from . import telemetry from .constants import ( DATA_DIR, DOCS_DEFAULT_LIMIT, @@ -162,18 +161,12 @@ def search(self, arguments: dict[str, Any] | None) -> list[ToolContent]: limit = _clamp_limit(arguments.get("limit"), self._default_limit) try: - results, embedding_duration_s = self._rank(query, limit) + results, _embedding_duration_s = self._rank(query, limit) except Exception as exc: - telemetry.record_search_docs(outcome="error", match_count=0) raise RuntimeError( "Documentation search embedding request failed. " "Check OPENAI_API_KEY and outbound connectivity to OpenAI." ) from exc - telemetry.record_search_docs( - outcome="success", - match_count=len(results), - embedding_duration_s=embedding_duration_s, - ) if not results: return [ diff --git a/src/mcp_server_appwrite/http_app.py b/src/mcp_server_appwrite/http_app.py index 1b9b435..3656e70 100644 --- a/src/mcp_server_appwrite/http_app.py +++ b/src/mcp_server_appwrite/http_app.py @@ -21,7 +21,6 @@ import logging import re import sys -import time from importlib.resources import files from mcp.server.auth.middleware.auth_context import AuthContextMiddleware @@ -155,16 +154,11 @@ async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None: user = scope.get("user") if not isinstance(user, AuthenticatedUser): - # A presented-but-invalid token was already counted (with its specific - # reason) by the token verifier. Only count the no-token case here so we - # don't double-count rejections. - if not _has_authorization_header(scope): - telemetry.record_auth(outcome="rejected", reason="missing") - else: - # A token was presented and rejected — this session never - # completes its MCP handshake. Unauthenticated discovery probes - # (no token at all) are part of the normal OAuth flow and are - # not counted as handshake failures. + # A token was presented and rejected — this session never completes + # its MCP handshake. Unauthenticated discovery probes (no token at + # all) are part of the normal OAuth flow and are not counted as + # handshake failures. + if _has_authorization_header(scope): telemetry.record_handshake_failure(reason="invalid_token") await _send_401(send) return @@ -280,7 +274,6 @@ def _bind_identity(self, scope: Scope, body: bytes) -> None: session_id=next(_connection_counter), client_name=client_info.get("name") or _client_from_user_agent(_header(scope, b"user-agent")), - client_version=client_info.get("version"), protocol_version=params.get("protocolVersion"), subject=subject, ) @@ -292,49 +285,6 @@ def _bind_identity(self, scope: Scope, body: bytes) -> None: ) -_KNOWN_HANDLERS = { - "/mcp", - "/healthz", - "/favicon.svg", - "/favicon.ico", - "/.well-known/oauth-protected-resource/mcp", -} - - -class HTTPMetricsMiddleware: - """Outermost ASGI wrapper emitting ``http.requests`` / ``http.request.duration``. - - The handler label is the raw path for the fixed route table above and - ``other`` for everything else, so unmatched scanner paths cannot explode - label cardinality.""" - - def __init__(self, app: ASGIApp) -> None: - self.app = app - - async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None: - if scope["type"] != "http": # pragma: no cover - await self.app(scope, receive, send) - return - - path = scope.get("path", "") - handler = path if path in _KNOWN_HANDLERS else "other" - method = scope.get("method", "GET") - start = time.monotonic() - status_holder = {"status": 500} - - async def send_wrapper(message) -> None: - if message["type"] == "http.response.start": - status_holder["status"] = message["status"] - await send(message) - - try: - await self.app(scope, receive, send_wrapper) - finally: - telemetry.record_http_request( - handler, method, status_holder["status"], time.monotonic() - start - ) - - async def protected_resource_metadata_endpoint(request: Request) -> JSONResponse: metadata = await protected_resource_metadata() headers = {**CORS_HEADERS, "Link": _icon_link_header()} @@ -400,7 +350,6 @@ async def lifespan(app: Starlette): ] middleware = [ - Middleware(HTTPMetricsMiddleware), Middleware( AuthenticationMiddleware, backend=BearerAuthBackend(AppwriteTokenVerifier()) ), diff --git a/src/mcp_server_appwrite/operator.py b/src/mcp_server_appwrite/operator.py index 8859b5c..f1647a7 100644 --- a/src/mcp_server_appwrite/operator.py +++ b/src/mcp_server_appwrite/operator.py @@ -314,15 +314,6 @@ def _get_context(self, arguments: dict[str, Any]) -> list[ToolContent]: if self._context_provider is None: raise RuntimeError("Appwrite context provider is not configured.") context = self._context_provider(arguments) - mode = "unknown" - if isinstance(context, dict): - connection = context.get("connection") - if isinstance(connection, dict): - mode = str(connection.get("mode", "unknown")) - include_services = bool( - arguments.get("include_services", arguments.get("includeServices", True)) - ) - telemetry.record_context_request(mode=mode, include_services=include_services) return [types.TextContent(type="text", text=json.dumps(context, indent=2))] def list_resources(self) -> list[types.Resource]: @@ -435,9 +426,6 @@ def _search_tools(self, arguments: dict[str, Any]) -> list[ToolContent]: include_mutating=include_mutating, limit=_normalize_limit(arguments.get("limit"), self._search_limit), ) - telemetry.record_search_tools( - include_mutating=include_mutating, match_count=len(matches) - ) lines: list[str] = [] if not matches: @@ -484,13 +472,10 @@ def _call_hidden_tool(self, raw_arguments: dict[str, Any]) -> list[ToolContent]: confirm_write = bool( raw_arguments.get("confirm_write", raw_arguments.get("confirmWrite", False)) ) - if entry.classification != "read": - if not confirm_write: - telemetry.record_write_confirmation(entry.classification, "blocked") - raise RuntimeError( - f"Tool {tool_name} is {entry.classification}. Re-run appwrite_call_tool with confirm_write=true if you intend to mutate Appwrite state." - ) - telemetry.record_write_confirmation(entry.classification, "confirmed") + if entry.classification != "read" and not confirm_write: + raise RuntimeError( + f"Tool {tool_name} is {entry.classification}. Re-run appwrite_call_tool with confirm_write=true if you intend to mutate Appwrite state." + ) project_id = raw_arguments.get("project_id", raw_arguments.get("projectId")) organization_id = raw_arguments.get( diff --git a/src/mcp_server_appwrite/server.py b/src/mcp_server_appwrite/server.py index a59171a..aa9fe6f 100644 --- a/src/mcp_server_appwrite/server.py +++ b/src/mcp_server_appwrite/server.py @@ -442,20 +442,17 @@ def _validate_fetch_url(url: str) -> None: """ parts = urlsplit(url) if parts.scheme not in ("http", "https"): - telemetry.record_upload_error("scheme") raise ValueError( f"Unsupported URL scheme '{parts.scheme}' — only http and https are allowed." ) host = parts.hostname if not host: - telemetry.record_upload_error("no_host") raise ValueError("URL is missing a host.") port = parts.port or (443 if parts.scheme == "https" else 80) try: infos = socket.getaddrinfo(host, port) except socket.gaierror as exc: - telemetry.record_upload_error("dns") raise ValueError(f"Could not resolve host '{host}'.") from exc for info in infos: @@ -468,7 +465,6 @@ def _validate_fetch_url(url: str) -> None: or ip.is_multicast or ip.is_unspecified ): - telemetry.record_upload_error("ssrf") raise ValueError( "Refusing to fetch a URL that resolves to a private, loopback, or " "link-local address." @@ -514,7 +510,6 @@ def _fetch_input_file(url: str, param_name: str) -> InputFile: declared = resp.headers.get("content-length") if declared is not None and declared.isdigit(): if int(declared) > MAX_FETCH_BYTES: - telemetry.record_upload_error("too_large") raise ValueError( f"File at URL for '{param_name}' is too large " f"({declared} bytes); max is {MAX_FETCH_BYTES} bytes." @@ -525,7 +520,6 @@ def _fetch_input_file(url: str, param_name: str) -> InputFile: for chunk in resp.iter_bytes(): total += len(chunk) if total > MAX_FETCH_BYTES: - telemetry.record_upload_error("too_large") raise ValueError( f"File at URL for '{param_name}' exceeds the max of " f"{MAX_FETCH_BYTES} bytes." @@ -538,13 +532,10 @@ def _fetch_input_file(url: str, param_name: str) -> InputFile: ) filename = _derive_filename(resp, url) except httpx.HTTPError as exc: - reason = "timeout" if isinstance(exc, httpx.TimeoutException) else "http_error" - telemetry.record_upload_error(reason) raise ValueError( f"Failed to fetch file from URL for '{param_name}': {exc}" ) from exc - telemetry.record_upload(source="url", outcome="success", size_bytes=len(data)) return InputFile.from_bytes(data, filename, mime_type or None) @@ -552,38 +543,32 @@ def _coerce_inline_content(value: Mapping, param_name: str) -> InputFile: filename = value.get("filename") content = value.get("content") if content is None: - telemetry.record_upload_error("decode") raise ValueError(f"Missing inline 'content' for '{param_name}'.") encoding = str(value.get("encoding", "utf-8")).lower() if encoding == "base64": try: data = base64.b64decode(content) except Exception as exc: - telemetry.record_upload_error("decode") raise ValueError(f"Invalid base64 content for '{param_name}'.") from exc elif encoding == "utf-8": data = str(content).encode("utf-8") else: - telemetry.record_upload_error("encoding") raise ValueError( f"Invalid encoding for '{param_name}'. Expected 'utf-8' or 'base64'." ) if len(data) > MAX_INLINE_BYTES: - telemetry.record_upload_error("too_large") raise ValueError( f"Inline content for '{param_name}' is too large " f"({len(data)} bytes, max {MAX_INLINE_BYTES}). For larger files pass " '{"url": "https://..."} so the server can download it directly.' ) - telemetry.record_upload(source="inline", outcome="success", size_bytes=len(data)) return InputFile.from_bytes(data, str(filename), value.get("mime_type")) def _coerce_path(path: str, param_name: str) -> InputFile: if _UPLOAD_TRANSPORT != "stdio": - telemetry.record_upload_error("path_unsupported") raise ValueError(HOSTED_PATH_GUIDANCE.format(param=param_name)) return InputFile.from_path(path) @@ -798,21 +783,9 @@ def execute_registered_tool( bound_method = getattr(service_cls(client), method_name) parsed = _parse_tool_name(name) - start = time.monotonic() try: result = bound_method(**prepared_arguments) except AppwriteException as exc: - telemetry.record_appwrite_call( - service=parsed["service_name"], - action=parsed["action_verb"], - classification=parsed["classification"], - outcome="error", - duration_s=time.monotonic() - start, - error_code=getattr(exc, "code", None), - error_type=getattr(exc, "type", None), - ) - if getattr(exc, "code", None) == 429: - telemetry.record_rate_limit(name) error_monitoring.capture_appwrite_exception( exc, service=parsed["service_name"], @@ -823,14 +796,6 @@ def execute_registered_tool( ) raise RuntimeError(_format_appwrite_error(exc)) from exc except Exception as exc: - telemetry.record_appwrite_call( - service=parsed["service_name"], - action=parsed["action_verb"], - classification=parsed["classification"], - outcome="error", - duration_s=time.monotonic() - start, - error_type="internal", - ) error_monitoring.capture_exception( exc, tags={ @@ -853,13 +818,6 @@ def execute_registered_tool( ) raise - telemetry.record_appwrite_call( - service=parsed["service_name"], - action=parsed["action_verb"], - classification=parsed["classification"], - outcome="success", - duration_s=time.monotonic() - start, - ) return _format_tool_result(name, result, prepared_arguments) @@ -1139,13 +1097,6 @@ async def handle_read_resource(uri) -> list[ReadResourceContents]: try: result = operator.read_resource(uri_str) except Exception as exc: - telemetry.record_resource_access( - resource_type, - outcome="error", - anomaly=( - "unknown_resource" if isinstance(exc, ValueError) else "read_error" - ), - ) telemetry.record_message( "resources/read", "error", @@ -1181,7 +1132,6 @@ async def handle_read_resource(uri) -> list[ReadResourceContents]: ) for item in result ) - telemetry.record_resource_access(resource_type, size_bytes=size_bytes) telemetry.record_message_size("sent", size_bytes) telemetry.record_message("resources/read", "success", time.monotonic() - start) return result diff --git a/src/mcp_server_appwrite/telemetry.py b/src/mcp_server_appwrite/telemetry.py index 9abcde2..70d7388 100644 --- a/src/mcp_server_appwrite/telemetry.py +++ b/src/mcp_server_appwrite/telemetry.py @@ -9,8 +9,10 @@ (grafana.com/grafana/dashboards/25252): after the standard OTLP-to-Prometheus translation (dots become underscores, unit and ``_total`` suffixes appended) the instruments below land as ``mcp_tool_calls_total``, -``mcp_tool_duration_seconds_bucket``, ``mcp_active_sessions``, -``http_requests_total``, and so on. +``mcp_tool_duration_seconds_bucket``, ``mcp_active_sessions``, and so on. +Only metrics queried by the hosted ``MCP`` Grafana dashboard +(appwrite-labs/dashboards ``mcp.json``) are emitted — when a panel is removed +there, drop its instrument here too. Design notes: @@ -22,9 +24,11 @@ telemetry must never break a request. * **Cardinality-disciplined.** No user id (``sub``), token, raw query text, file name, result id, or IP is ever used as a metric attribute. ``client_id`` is the - MCP client *name* (``claude-code``, ``cursor``, ...), a small bounded set. - Distinct-user and per-client session counts are derived in-process from rolling - TTL sets and exposed only as aggregate gauges. + MCP client *name* (``claude-code``, ``cursor``, ...) checked against a fixed + allowlist of known clients — anything else is labeled ``other``, so a client + cannot mint arbitrary label values. Distinct-user and per-client session counts + are derived in-process from rolling TTL sets and exposed only as aggregate + gauges. * **Stateless sessions.** The hosted transport is stateless — every HTTP request is its own short-lived MCP session. "Session" metrics therefore describe user activity windows: a session starts when a (client, user) pair is first seen and @@ -51,11 +55,10 @@ from contextvars import ContextVar from typing import Any, Iterable -from .constants import ACTIVE_WINDOW_SECONDS +from .constants import ACTIVE_WINDOW_SECONDS, KNOWN_MCP_CLIENTS _enabled = False _lock = threading.Lock() -_transport = "http" # Metric instruments, populated by init_telemetry when enabled. _instruments: dict[str, Any] = {} @@ -68,8 +71,6 @@ _active_sessions: dict[tuple[str, str], list[float]] = {} # (protocol version, client, subject) -> expiry _active_versions: dict[tuple[str, str, str], float] = {} -# subject -> expiry; users that recently hit an Appwrite rate limit (429) -_rate_limited: dict[str, float] = {} _active_lock = threading.Lock() # Dedupe connection events: one per MCP session object. @@ -78,9 +79,6 @@ # Identity of the request currently being served, set once per request by the # MCP handlers and read by every record helper that labels by client. _request_client: ContextVar[str] = ContextVar("appwrite_mcp_client", default="unknown") -_request_subject: ContextVar[str | None] = ContextVar( - "appwrite_mcp_subject", default=None -) # CPU gauge state: previous (wall clock, process cpu time) sample. _cpu_sample: list[float] = [] @@ -198,15 +196,7 @@ def _histogram( def _build_instruments(meter: Any, transport: str, version: str) -> None: - global _transport - _transport = "streamable-http" if transport == "http" else transport - # --- Transport & sessions ------------------------------------------------ - _instruments["connection"] = meter.create_counter( - "mcp.connection", - unit="{connection}", - description="New MCP sessions (initialize handshakes) by transport.", - ) _instruments["handshake"] = meter.create_counter( "mcp.handshake", unit="{handshake}", @@ -303,128 +293,6 @@ def _build_instruments(meter: Any, transport: str, version: str) -> None: boundaries=_TOKEN_BUCKETS, ) - # --- Rate limiting --------------------------------------------------------- - _instruments["rate_limit"] = meter.create_counter( - "mcp.rate.limit", - unit="{event}", - description="Appwrite API rate-limit (HTTP 429) responses surfaced to tools.", - ) - - # --- Resource access -------------------------------------------------------- - _instruments["resource_accessed"] = meter.create_counter( - "mcp.resource.accessed", - unit="{read}", - description="resources/read calls by resource type.", - ) - _instruments["resource_errors"] = meter.create_counter( - "mcp.resource.errors", - unit="{error}", - description="Failed resources/read calls by resource type.", - ) - _instruments["resource_size"] = _histogram( - meter, - "mcp.resource.size", - unit="By", - description="Resource payload size served by resources/read.", - boundaries=_BYTE_BUCKETS, - ) - _instruments["resource_anomaly"] = meter.create_counter( - "mcp.resource.access.anomaly", - unit="{event}", - description="Suspicious resource reads (unknown or expired URIs).", - ) - - # --- HTTP layer ---------------------------------------------------------------- - _instruments["http_requests"] = meter.create_counter( - "http.requests", - unit="{request}", - description="HTTP requests served by the hosted transport, by handler.", - ) - _instruments["http_request_duration"] = _histogram( - meter, - "http.request.duration", - unit="s", - description="HTTP request duration by handler.", - ) - - # --- Appwrite-specific operator metrics (not on the reference dashboard) ----- - _instruments["appwrite_calls"] = meter.create_counter( - "mcp.appwrite.calls", - unit="{call}", - description="Hidden Appwrite catalog tool executions.", - ) - _instruments["appwrite_call_duration"] = _histogram( - meter, - "mcp.appwrite.call.duration", - unit="s", - description="Underlying Appwrite REST call duration.", - ) - _instruments["appwrite_errors"] = meter.create_counter( - "mcp.appwrite.errors", - unit="{error}", - description="Failed Appwrite catalog tool executions.", - ) - _instruments["write_confirmations"] = meter.create_counter( - "mcp.write.confirmations", - unit="{confirmation}", - description="Write/delete confirmation outcomes (confirmed vs blocked).", - ) - _instruments["search_tools_queries"] = meter.create_counter( - "mcp.search_tools.queries", - unit="{query}", - description="appwrite_search_tools catalog searches.", - ) - _instruments["search_tools_results"] = _histogram( - meter, - "mcp.search_tools.results", - unit="{match}", - description="Match count returned by appwrite_search_tools.", - ) - _instruments["search_docs_queries"] = meter.create_counter( - "mcp.search_docs.queries", - unit="{query}", - description="appwrite_search_docs documentation searches.", - ) - _instruments["search_docs_embedding_duration"] = _histogram( - meter, - "mcp.search_docs.embedding.duration", - unit="s", - description="Query embedding duration for docs search.", - ) - _instruments["context_requests"] = meter.create_counter( - "mcp.context.requests", - unit="{request}", - description="appwrite_get_context invocations.", - ) - _instruments["auth_validations"] = meter.create_counter( - "mcp.auth.validations", - unit="{validation}", - description="Bearer-token validation outcomes.", - ) - _instruments["auth_duration"] = _histogram( - meter, - "mcp.auth.duration", - unit="s", - description="Bearer-token verification duration.", - ) - _instruments["uploads"] = meter.create_counter( - "mcp.uploads", - unit="{upload}", - description="File upload attempts.", - ) - _instruments["upload_bytes"] = _histogram( - meter, - "mcp.upload.bytes", - unit="By", - description="Uploaded file size.", - boundaries=_BYTE_BUCKETS, - ) - _instruments["upload_errors"] = meter.create_counter( - "mcp.upload.errors", - unit="{error}", - description="File upload failures.", - ) - # --- Observable gauges ------------------------------------------------------ meter.create_observable_gauge( "mcp.active_sessions", @@ -444,12 +312,6 @@ def _build_instruments(meter: Any, transport: str, version: str) -> None: unit="{session}", description="Active sessions by negotiated MCP protocol version and client.", ) - meter.create_observable_gauge( - "mcp.rate.limit.active", - callbacks=[_observe_rate_limited], - unit="{user}", - description="Distinct users rate-limited by Appwrite in the last 5 minutes.", - ) meter.create_observable_gauge( "mcp.cpu.usage.percent", callbacks=[_observe_cpu], @@ -478,16 +340,25 @@ def _observe_info(_options: Any): _CLIENT_NAME_WHITESPACE = re.compile(r"\s+") +_KNOWN_CLIENTS_BY_LENGTH = sorted(KNOWN_MCP_CLIENTS, key=len, reverse=True) + def _normalize_client_name(name: str | None) -> str | None: - """Canonicalize a client-reported name so one client maps to one - ``client_id`` label value. ``clientInfo.name`` arrives raw from the - initialize request while User-Agent-derived names are lowercased product - tokens, so the same client would otherwise split (``Trae`` vs ``trae``).""" + """Canonicalize a client-reported name to a bounded ``client_id`` label + value: lowercased with whitespace collapsed (``clientInfo.name`` arrives raw + from the initialize request while User-Agent-derived names are lowercased + product tokens, so the same client would otherwise split — ``Trae`` vs + ``trae``), then matched against the known-client allowlist. Unrecognized + names collapse to ``other``.""" if not name: return None text = _CLIENT_NAME_WHITESPACE.sub("-", str(name).strip().lower())[:64] - return text or None + if not text: + return None + for known in _KNOWN_CLIENTS_BY_LENGTH: + if text == known or text.startswith(known + "-"): + return known + return "other" def set_request_identity( @@ -503,7 +374,6 @@ def set_request_identity( # HTTP-layer middleware) to "unknown". client = _normalize_client_name(client_name) or _request_client.get() _request_client.set(client) - _request_subject.set(subject) if not _enabled: # The rolling stores are only pruned by the gauge callbacks, which never # run while disabled — do not let them grow. @@ -587,16 +457,6 @@ def _observe_protocol_versions(_options: Any) -> Iterable[Any]: ] -def _observe_rate_limited(_options: Any) -> Iterable[Any]: - from opentelemetry.metrics import Observation - - now = time.monotonic() - with _active_lock: - _prune(_rate_limited, now) - count = len(_rate_limited) - return [Observation(count)] - - def _observe_cpu(_options: Any) -> Iterable[Any]: from opentelemetry.metrics import Observation @@ -683,12 +543,11 @@ def record_connection( *, session_id: int, client_name: str | None, - client_version: str | None, protocol_version: str | None, subject: str | None, ) -> None: - """Count a new MCP session (connection + successful handshake), deduped per - session object, and refresh the request-identity stores.""" + """Count a successful MCP handshake, deduped per session object, and refresh + the request-identity stores.""" set_request_identity( client_name=client_name, subject=subject, @@ -705,15 +564,6 @@ def record_connection( _seen_sessions.clear() _seen_sessions.add(session_id) client = _normalize_client_name(client_name) or "unknown" - _safe_add( - "connection", - 1, - { - "transport": _transport, - "client_id": client, - "client_version": client_version, - }, - ) _safe_add("handshake", 1, {"status": "success", "client_id": client}) @@ -824,152 +674,3 @@ def record_hallucination(attempted_tool: Any) -> None: "client_id": current_client_id(), }, ) - - -def record_rate_limit(tool_name: str) -> None: - _safe_add( - "rate_limit", - 1, - {"tool_name": tool_name, "client_id": current_client_id()}, - ) - subject = _request_subject.get() - if not _enabled or not subject: - return - with _active_lock: - _rate_limited[subject] = time.monotonic() + ACTIVE_WINDOW_SECONDS - - -# --- Resource access ----------------------------------------------------------------- - - -def record_resource_access( - resource: str, - *, - outcome: str = "success", - size_bytes: int | None = None, - anomaly: str | None = None, -) -> None: - _safe_add( - "resource_accessed", - 1, - {"resource": resource, "client_id": current_client_id()}, - ) - if outcome == "error": - _safe_add("resource_errors", 1, {"resource": resource}) - if size_bytes is not None: - _safe_record("resource_size", size_bytes, {"resource": resource}) - if anomaly: - _safe_add("resource_anomaly", 1, {"anomaly_type": anomaly}) - - -# --- HTTP layer ------------------------------------------------------------------------- - - -def record_http_request( - handler: str, method: str, status_code: int, duration_s: float -) -> None: - _safe_add( - "http_requests", - 1, - {"handler": handler, "method": method, "code": str(status_code)}, - ) - _safe_record("http_request_duration", duration_s, {"handler": handler}) - - -# --- Appwrite-specific helpers (unchanged surface) ----------------------------------- - - -def record_appwrite_call( - *, - service: str, - action: str, - classification: str, - outcome: str, - duration_s: float, - error_code: Any = None, - error_type: str | None = None, -) -> None: - attrs = { - "appwrite.service": service or "unknown", - "appwrite.action": action or "unknown", - "appwrite.classification": classification or "unknown", - "outcome": outcome, - } - _safe_add("appwrite_calls", 1, attrs) - _safe_record( - "appwrite_call_duration", - duration_s, - { - "appwrite.service": service or "unknown", - "appwrite.action": action or "unknown", - }, - ) - if outcome == "error": - _safe_add( - "appwrite_errors", - 1, - { - "appwrite.service": service or "unknown", - "appwrite.action": action or "unknown", - "error.code": str(error_code) if error_code is not None else "unknown", - "error.type": error_type or "unknown", - }, - ) - - -def record_write_confirmation(classification: str, outcome: str) -> None: - _safe_add( - "write_confirmations", - 1, - {"appwrite.classification": classification, "outcome": outcome}, - ) - - -def record_search_tools(*, include_mutating: bool, match_count: int) -> None: - _safe_add( - "search_tools_queries", - 1, - {"include_mutating": include_mutating, "matched": match_count > 0}, - ) - _safe_record("search_tools_results", match_count, {}) - - -def record_search_docs( - *, outcome: str, match_count: int, embedding_duration_s: float | None = None -) -> None: - _safe_add( - "search_docs_queries", 1, {"outcome": outcome, "matched": match_count > 0} - ) - if embedding_duration_s is not None: - _safe_record( - "search_docs_embedding_duration", embedding_duration_s, {"outcome": outcome} - ) - - -def record_context_request(*, mode: str, include_services: bool) -> None: - _safe_add( - "context_requests", 1, {"mode": mode, "include_services": include_services} - ) - - -def record_auth( - *, - outcome: str, - reason: str | None = None, - duration_s: float | None = None, - count: bool = True, -) -> None: - if count: - _safe_add("auth_validations", 1, {"outcome": outcome, "reason": reason}) - if duration_s is not None: - _safe_record("auth_duration", duration_s, {"outcome": outcome}) - - -def record_upload(*, source: str, outcome: str, size_bytes: int | None = None) -> None: - _safe_add("uploads", 1, {"source": source, "outcome": outcome}) - if outcome == "success" and size_bytes is not None: - _safe_record("upload_bytes", size_bytes, {"source": source}) - - -def record_upload_error(reason: str) -> None: - _safe_add("upload_errors", 1, {"reason": reason}) diff --git a/tests/unit/test_http_app.py b/tests/unit/test_http_app.py index be09952..199805a 100644 --- a/tests/unit/test_http_app.py +++ b/tests/unit/test_http_app.py @@ -262,7 +262,7 @@ def _points(self, metric_name: str) -> list: return list(metric.data.data_points) return [] - def test_initialize_records_connection_and_replays_body(self): + def test_initialize_records_handshake_and_replays_body(self): body = json.dumps( { "jsonrpc": "2.0", @@ -276,10 +276,9 @@ def test_initialize_records_connection_and_replays_body(self): ).encode() replayed = self._run(body, [(b"user-agent", b"claude-code/2.0")]) self.assertEqual(replayed, [body]) - connections = self._points("mcp.connection") - self.assertEqual(len(connections), 1) - self.assertEqual(connections[0].attributes.get("client_id"), "claude-code") handshakes = self._points("mcp.handshake") + self.assertEqual(len(handshakes), 1) + self.assertEqual(handshakes[0].attributes.get("client_id"), "claude-code") self.assertEqual(handshakes[0].attributes.get("status"), "success") def test_tool_call_binds_identity_from_headers(self): @@ -292,8 +291,8 @@ def test_tool_call_binds_identity_from_headers(self): ], ) self.assertEqual(self.seen_client, ["cursor"]) - # No connection counted for non-initialize requests. - self.assertEqual(self._points("mcp.connection"), []) + # No handshake counted for non-initialize requests. + self.assertEqual(self._points("mcp.handshake"), []) if __name__ == "__main__": diff --git a/tests/unit/test_telemetry.py b/tests/unit/test_telemetry.py index a92d335..d676212 100644 --- a/tests/unit/test_telemetry.py +++ b/tests/unit/test_telemetry.py @@ -47,7 +47,6 @@ def _clear_stores(self) -> None: telemetry._active_users.clear() telemetry._active_sessions.clear() telemetry._active_versions.clear() - telemetry._rate_limited.clear() telemetry._seen_sessions.clear() def points(self, metric_name: str) -> list: @@ -68,71 +67,21 @@ def connect(self, session_id=1, client="claude-code", subject="user-a"): telemetry.record_connection( session_id=session_id, client_name=client, - client_version="1.0", protocol_version="2025-06-18", subject=subject, ) -class RecordHelperTests(TelemetryHarness): - def test_appwrite_call_success_labels(self): - telemetry.record_appwrite_call( - service="storage", - action="create", - classification="write", - outcome="success", - duration_s=0.05, - ) - points = self.points("mcp.appwrite.calls") - self.assertEqual(len(points), 1) - self.assertEqual(points[0].value, 1) - self.assertAttr(points[0], "appwrite.service", "storage") - self.assertAttr(points[0], "appwrite.action", "create") - self.assertAttr(points[0], "appwrite.classification", "write") - self.assertAttr(points[0], "outcome", "success") - # No error counter on success. - self.assertEqual(self.points("mcp.appwrite.errors"), []) - - def test_appwrite_call_error_emits_error_counter(self): - telemetry.record_appwrite_call( - service="users", - action="get", - classification="read", - outcome="error", - duration_s=0.01, - error_code=404, - error_type="user_not_found", - ) - errors = self.points("mcp.appwrite.errors") - self.assertEqual(len(errors), 1) - self.assertAttr(errors[0], "appwrite.service", "users") - self.assertAttr(errors[0], "error.code", "404") - self.assertAttr(errors[0], "error.type", "user_not_found") - - def test_auth_rejected_then_duration_without_double_count(self): - # _verify_sync-style: counter with reason, no duration. - telemetry.record_auth(outcome="rejected", reason="signature") - # verify_token-style: duration only, no counter. - telemetry.record_auth(outcome="rejected", duration_s=0.02, count=False) - validations = self.points("mcp.auth.validations") - self.assertEqual(len(validations), 1) - self.assertEqual(validations[0].value, 1) - self.assertAttr(validations[0], "reason", "signature") - - class SessionTests(TelemetryHarness): def test_active_sessions_counts_distinct_subjects(self): self.connect(session_id=1, subject="user-a") self.connect(session_id=2, subject="user-b") sessions = self.points("mcp.active_sessions") self.assertEqual(sessions[0].value, 2) - connections = self.points("mcp.connection") - self.assertEqual(sum(p.value for p in connections), 2) - self.assertAttr(connections[0], "client_id", "claude-code") - self.assertAttr(connections[0], "transport", "streamable-http") handshakes = self.points("mcp.handshake") self.assertEqual(sum(p.value for p in handshakes), 2) self.assertAttr(handshakes[0], "status", "success") + self.assertAttr(handshakes[0], "client_id", "claude-code") def test_active_sessions_by_client_and_protocol_version(self): self.connect(session_id=1, client="claude-code", subject="user-a") @@ -144,11 +93,11 @@ def test_active_sessions_by_client_and_protocol_version(self): self.assertEqual(sum(p.value for p in versions), 2) self.assertAttr(versions[0], "version", "2025-06-18") - def test_connection_deduped_per_session(self): + def test_handshake_deduped_per_session(self): for _ in range(3): self.connect(session_id=42, client="cursor", subject="user-c") - connections = self.points("mcp.connection") - self.assertEqual(sum(p.value for p in connections), 1) + handshakes = self.points("mcp.handshake") + self.assertEqual(sum(p.value for p in handshakes), 1) def test_expired_session_records_duration_and_idle_disconnect(self): self.connect(session_id=1, client="cursor", subject="user-a") @@ -181,8 +130,8 @@ def test_client_names_are_case_normalized(self): by_client = self.points("mcp.active_sessions.by_client") counts = {p.attributes["client_id"]: p.value for p in by_client} self.assertEqual(counts, {"trae": 1}) - connections = self.points("mcp.connection") - self.assertAttr(connections[0], "client_id", "trae") + handshakes = self.points("mcp.handshake") + self.assertAttr(handshakes[0], "client_id", "trae") def test_client_name_normalization_shapes(self): cases = { @@ -192,13 +141,35 @@ def test_client_name_normalization_shapes(self): "": None, None: None, " ": None, - "X" * 100: "x" * 64, } for raw, expected in cases.items(): self.assertEqual( telemetry._normalize_client_name(raw), expected, msg=repr(raw) ) + def test_unknown_client_names_collapse_to_other(self): + # client_id comes from client-controlled input; anything outside the + # known-client allowlist must not mint a new label value. + cases = { + "X" * 100: "other", + "my-evil-client-123": "other", + "definitely not a real client": "other", + # Prefixed variants map to the known client, longest match first. + "claude-code-nightly": "claude-code", + "Cursor Nightly": "cursor", + "claude": "claude", + } + for raw, expected in cases.items(): + self.assertEqual( + telemetry._normalize_client_name(raw), expected, msg=repr(raw) + ) + + def test_unknown_client_sessions_are_labeled_other(self): + self.connect(session_id=1, client="totally-made-up-agent", subject="user-a") + by_client = self.points("mcp.active_sessions.by_client") + counts = {p.attributes["client_id"]: p.value for p in by_client} + self.assertEqual(counts, {"other": 1}) + class MessageTests(TelemetryHarness): def test_message_success(self): @@ -283,45 +254,8 @@ def test_hallucination_sanitizes_tool_name(self): self.assertNotIn(" ", attempted) self.assertNotIn("!", attempted) - def test_rate_limit_marks_active_subject(self): - self.connect(subject="user-a") - telemetry.record_rate_limit("tables_db_create") - events = self.points("mcp.rate.limit") - self.assertAttr(events[0], "tool_name", "tables_db_create") - active = self.points("mcp.rate.limit.active") - self.assertEqual(active[0].value, 1) - - -class ResourceTests(TelemetryHarness): - def test_resource_access_success_records_size(self): - self.connect() - telemetry.record_resource_access("result", size_bytes=2048) - accessed = self.points("mcp.resource.accessed") - self.assertAttr(accessed[0], "resource", "result") - sizes = self.points("mcp.resource.size") - self.assertEqual(sizes[0].sum, 2048) - self.assertEqual(self.points("mcp.resource.errors"), []) - - def test_resource_access_error_records_anomaly(self): - telemetry.record_resource_access( - "result", outcome="error", anomaly="unknown_resource" - ) - errors = self.points("mcp.resource.errors") - self.assertEqual(len(errors), 1) - anomalies = self.points("mcp.resource.access.anomaly") - self.assertAttr(anomalies[0], "anomaly_type", "unknown_resource") - - -class HttpTests(TelemetryHarness): - def test_http_request(self): - telemetry.record_http_request("/mcp", "POST", 200, 0.02) - requests = self.points("http.requests") - self.assertAttr(requests[0], "handler", "/mcp") - self.assertAttr(requests[0], "method", "POST") - self.assertAttr(requests[0], "code", "200") - durations = self.points("http.request.duration") - self.assertEqual(durations[0].count, 1) +class SystemGaugeTests(TelemetryHarness): def test_system_gauges_registered(self): # CPU needs two observations for a delta; memory should report on the first. self.reader.get_metrics_data() @@ -352,36 +286,16 @@ def make_runtime(self, executor): } return Operator(manager, executor) - def test_write_confirmation_blocked(self): + def test_blocked_write_counts_tool_error(self): runtime = self.make_runtime(lambda name, arguments, *_: []) with self.assertRaises(RuntimeError): runtime.execute_public_tool( "appwrite_call_tool", {"tool_name": "tables_db_create", "arguments": {"database_id": "db"}}, ) - confirmations = self.points("mcp.write.confirmations") - self.assertEqual(len(confirmations), 1) - self.assertAttr(confirmations[0], "outcome", "blocked") - self.assertAttr(confirmations[0], "appwrite.classification", "write") - - def test_write_confirmation_confirmed(self): - runtime = self.make_runtime( - lambda name, arguments, *_: [types.TextContent(type="text", text="ok")] - ) - runtime.execute_public_tool( - "appwrite_call_tool", - { - "tool_name": "tables_db_create", - "confirm_write": True, - "database_id": "db", - }, - ) - confirmed = [ - p - for p in self.points("mcp.write.confirmations") - if p.attributes.get("outcome") == "confirmed" - ] - self.assertEqual(len(confirmed), 1) + errors = self.points("mcp.tool.errors") + self.assertEqual(len(errors), 1) + self.assertAttr(errors[0], "error_type", "RuntimeError") def test_tool_call_counter(self): runtime = self.make_runtime( @@ -421,14 +335,8 @@ def test_no_emission_when_disabled(self): telemetry._enabled = False try: telemetry.record_message("tools/call", "success", 0.01) - telemetry.record_appwrite_call( - service="storage", - action="create", - classification="write", - outcome="success", - duration_s=0.01, - ) - telemetry.record_auth(outcome="rejected", reason="malformed") + telemetry.record_tool_call("appwrite_call_tool", "success", 0.01) + telemetry.record_handshake_failure(reason="invalid_token") recorded = { metric.name @@ -438,8 +346,8 @@ def test_no_emission_when_disabled(self): if metric.data.data_points } self.assertNotIn("mcp.messages.received", recorded) - self.assertNotIn("mcp.appwrite.calls", recorded) - self.assertNotIn("mcp.auth.validations", recorded) + self.assertNotIn("mcp.tool.calls", recorded) + self.assertNotIn("mcp.handshake", recorded) finally: telemetry._instruments.clear() @@ -457,7 +365,6 @@ def test_record_connection_does_not_grow_stores_when_disabled(self): telemetry.record_connection( session_id=7, client_name="claude", - client_version="1.0", protocol_version="2025-06-18", subject="user-x", )