Merge pull request #50 from appwrite/feat/opentelemetry-metrics

feat: add OpenTelemetry metrics for the hosted MCP server

Author: ChiragAgg5k

AGENTS.md

@@ -30,10 +30,19 @@ Source lives in `src/mcp_server_appwrite/`:
 | `operator.py` | The compact "operator" surface — `appwrite_search_tools`, `appwrite_call_tool`, result/resource storage, write confirmation. |
 | `context.py` | `appwrite_get_context` — workspace summary (project, services, account/org for OAuth). |
 | `docs_search.py` | In-process semantic docs search (`appwrite_search_docs`) over a prebuilt index. |
+| `telemetry.py` | OpenTelemetry metrics layer (OTLP/HTTP). No-op unless an OTLP endpoint is configured and the transport is `http`. |
 | `data/` | Committed docs index artifact (`docs_index.npz`, `docs_index_meta.json`), shipped in the wheel/image. |
 
 `scripts/build_docs_index.py` rebuilds the docs index (requires `OPENAI_API_KEY`).
 
+### Telemetry (metrics)
+
+`telemetry.py` emits OpenTelemetry metrics (prefixed `mcp.`) over OTLP/HTTP. It is a
+no-op unless the transport is `http` and `OTEL_EXPORTER_OTLP_ENDPOINT` is set, so
+`stdio` and unconfigured servers stay silent. In the cluster that endpoint is the
+Alloy collector, which adds the `deployment.*` labels and forwards upstream — the app
+needs no credentials. Dashboards live in the `dashboards` repo under `MCP/`.
+
 ### Tool surface (key design point)
 
 The server boots in a compact workflow: the client sees up to 4 tools

compose.yaml

@@ -15,6 +15,7 @@ services:
       OPENAI_API_KEY: ${OPENAI_API_KEY:-}
       DOCS_SEARCH_MIN_SCORE: ${DOCS_SEARCH_MIN_SCORE:-0.25}
       DOCS_SEARCH_LIMIT: ${DOCS_SEARCH_LIMIT:-5}
+      OTEL_EXPORTER_OTLP_ENDPOINT: ${OTEL_EXPORTER_OTLP_ENDPOINT:-}
     healthcheck:
       test:
         [

pyproject.toml

@@ -15,6 +15,9 @@ dependencies = [
     "pyjwt[crypto]>=2.9.0",
     "numpy>=2.0",
     "openai>=1.40",
+    "opentelemetry-api>=1.27",
+    "opentelemetry-sdk>=1.27",
+    "opentelemetry-exporter-otlp-proto-http>=1.27",
 ]
 
 [project.optional-dependencies]

src/mcp_server_appwrite/auth.py

@@ -25,6 +25,8 @@
 from jwt import PyJWKClient
 from mcp.server.auth.provider import AccessToken, TokenVerifier
 
+from . import telemetry
+
 DEFAULT_ENDPOINT = "https://cloud.appwrite.io/v1"
 DEFAULT_PROJECT_ID = "console"
 
@@ -198,13 +200,15 @@ 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")
         try:
             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"]
@@ -213,17 +217,20 @@ 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:
@@ -238,10 +245,12 @@ 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 ""
@@ -276,9 +285,18 @@ 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 anyio.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

src/mcp_server_appwrite/docs_search.py

@@ -15,11 +15,14 @@
 
 import json
 import os
+import time
 from pathlib import Path
 from typing import Any, Callable
 
 import mcp.types as types
 
+from . import telemetry
+
 ToolContent = types.TextContent | types.ImageContent | types.EmbeddedResource
 
 TOOL_NAME = "appwrite_search_docs"
@@ -155,7 +158,12 @@ def search(self, arguments: dict[str, Any] | None) -> list[ToolContent]:
             )
 
         limit = _clamp_limit(arguments.get("limit"), self._default_limit)
-        results = self._rank(query, limit)
+        results, embedding_duration_s = self._rank(query, limit)
+        telemetry.record_search_docs(
+            outcome="success",
+            match_count=len(results),
+            embedding_duration_s=embedding_duration_s,
+        )
 
         if not results:
             return [
@@ -173,13 +181,16 @@ def search(self, arguments: dict[str, Any] | None) -> list[ToolContent]:
             )
         ]
 
-    def _rank(self, query: str, limit: int) -> list[dict[str, Any]]:
+    def _rank(self, query: str, limit: int) -> tuple[list[dict[str, Any]], float]:
+        """Return the ranked pages and the embedding call's duration in seconds."""
         import numpy as np
 
+        embed_start = time.monotonic()
         embedding = np.asarray(self._embedder(query), dtype=np.float32)
+        embedding_duration_s = time.monotonic() - embed_start
         norm = float(np.linalg.norm(embedding))
         if norm == 0.0:
-            return []
+            return [], embedding_duration_s
         embedding /= norm
 
         scores = self._vectors @ embedding  # cosine similarity (both normalized)
@@ -207,4 +218,4 @@ def _rank(self, query: str, limit: int) -> list[dict[str, Any]]:
                     "content": page.get("content", ""),
                 }
             )
-        return results
+        return results, embedding_duration_s

src/mcp_server_appwrite/http_app.py

@@ -31,6 +31,7 @@
 from starlette.routing import Route
 from starlette.types import Receive, Scope, Send
 
+from . import telemetry
 from .auth import (
     AppwriteTokenVerifier,
     protected_resource_metadata,
@@ -111,6 +112,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")
             await _send_401(send)
             return
 
@@ -122,6 +128,13 @@ async def _preflight(self, send: Send) -> None:
         await send({"type": "http.response.body", "body": b""})
 
 
+def _has_authorization_header(scope: Scope) -> bool:
+    for name, _value in scope.get("headers", []):
+        if name == b"authorization":
+            return True
+    return False
+
+
 async def protected_resource_metadata_endpoint(request: Request) -> JSONResponse:
     metadata = await protected_resource_metadata()
     return JSONResponse(metadata, headers=_CORS_HEADERS)
@@ -132,6 +145,7 @@ async def health_endpoint(request: Request) -> PlainTextResponse:
 
 
 def build_app() -> Starlette:
+    telemetry.init_telemetry("http", SERVER_VERSION)
     tools_manager = build_catalog_tools_manager()
     operator = build_operator(tools_manager)
     server = build_mcp_server(operator, transport="http")

src/mcp_server_appwrite/operator.py

@@ -2,6 +2,7 @@
 
 import json
 import re
+import time
 from collections import OrderedDict
 from dataclasses import dataclass
 from typing import Any, Callable
@@ -11,6 +12,7 @@
 import mcp.types as types
 from mcp.server.lowlevel.helper_types import ReadResourceContents
 
+from . import telemetry
 from .docs_search import DocsSearch
 from .tool_manager import ToolManager
 
@@ -253,6 +255,19 @@ def has_public_tool(self, name: str) -> bool:
 
     def execute_public_tool(
         self, name: str, arguments: dict[str, Any] | None
+    ) -> list[ToolContent]:
+        start = time.monotonic()
+        outcome = "success"
+        try:
+            return self._dispatch_public_tool(name, arguments)
+        except Exception:
+            outcome = "error"
+            raise
+        finally:
+            telemetry.record_tool_call(name, outcome, time.monotonic() - start)
+
+    def _dispatch_public_tool(
+        self, name: str, arguments: dict[str, Any] | None
     ) -> list[ToolContent]:
         if name == "appwrite_get_context":
             return self._get_context(arguments or {})
@@ -269,6 +284,15 @@ 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]:
@@ -381,6 +405,9 @@ 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:
@@ -426,10 +453,13 @@ 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" 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."
-            )
+        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")
 
         project_id = raw_arguments.get("project_id", raw_arguments.get("projectId"))
         organization_id = raw_arguments.get(
@@ -454,6 +484,7 @@ def _preview_or_store_result(
             stored_result = self._result_store.save(
                 tool_name, content, _serialize_content(content)
             )
+            telemetry.record_result_stored(tool_name)
             preview = full_text[: self._preview_threshold]
             return [
                 types.TextContent(
@@ -468,6 +499,7 @@ def _preview_or_store_result(
         stored_result = self._result_store.save(
             tool_name, content, _serialize_content(content)
         )
+        telemetry.record_result_stored(tool_name)
         summary = ", ".join(_summarize_content_item(item) for item in content)
         return [
             types.TextContent(