diff --git a/docs/mcp.md b/docs/mcp.md index dfdc12cf..7d82be2f 100644 --- a/docs/mcp.md +++ b/docs/mcp.md @@ -251,7 +251,7 @@ Use `MCPVerifier` directly when you're not using FastMCP — works with the raw ```python from tenuo import Authorizer, PublicKey, CompiledMcpConfig, McpConfig -from tenuo.mcp import MCPVerifier +from tenuo.mcp import MCPVerificationResult, MCPVerifier authorizer = Authorizer(trusted_roots=[PublicKey.from_bytes(root_pub)]) config = CompiledMcpConfig.compile(McpConfig.from_file("mcp-config.yaml")) @@ -266,6 +266,35 @@ execute_tool(result.clean_arguments) clean = verifier.verify_or_raise("read_file", {"path": path}, meta=request_meta) ``` +To record end-to-end verification latency in your own metrics pipeline, pass a +`latency_observer`. It receives a result snapshot and elapsed time in +microseconds on every allow and denial path. Observer failures and snapshot +mutations never change the authorization decision. + +```python +from prometheus_client import Histogram + +verification_latency = Histogram( + "tenuo_mcp_verification_latency_seconds", + "End-to-end MCP warrant verification latency", + ["tool", "allowed"], +) + + +def observe_latency(result: MCPVerificationResult, latency_us: int) -> None: + verification_latency.labels( + tool=result.tool, + allowed=str(result.allowed).lower(), + ).observe(latency_us / 1_000_000) + + +verifier = MCPVerifier( + authorizer=authorizer, + config=config, + latency_observer=observe_latency, +) +``` + ### Pattern 4: Securing LangChain MCP Adapters If you're already using `langchain-mcp-adapters`, wrap its tools with `guard_tools()`: diff --git a/tenuo-python/tenuo/mcp/server.py b/tenuo-python/tenuo/mcp/server.py index 24fe25fb..e06233ff 100644 --- a/tenuo-python/tenuo/mcp/server.py +++ b/tenuo-python/tenuo/mcp/server.py @@ -104,8 +104,10 @@ async def call_tool(name: str, arguments: dict) -> list: import base64 import logging +import time +from copy import deepcopy from dataclasses import dataclass, field -from typing import Any, Dict, List, Optional +from typing import Any, Callable, Dict, List, Optional from .._pop_canonicalize import strip_none_values from ..exceptions import ( @@ -375,6 +377,9 @@ def __init__( require_warrant: bool = True, control_plane: Optional[Any] = None, nonce_store: Optional[Any] = None, + latency_observer: Callable[ + [MCPVerificationResult, int], None + ] | None = None, ) -> None: """ Args: @@ -399,6 +404,12 @@ def __init__( ``enable_default_nonce_store()`` at startup or pass an explicit ``NonceStore(backend=RedisNonceBackend(...))`` for distributed deployments. + latency_observer: Optional callback invoked after every verification + with a snapshot of the final :class:`MCPVerificationResult` and + end-to-end elapsed time in microseconds. Observer failures and + snapshot mutations never change the authorization result. Use + this to feed local Prometheus, OpenTelemetry, or structured-log + instrumentation. """ from tenuo._extension import require_extension require_extension("MCPVerifier") @@ -411,6 +422,7 @@ def __init__( control_plane = get_or_create() self._control_plane = control_plane self._nonce_store = nonce_store + self._latency_observer = latency_observer def verify( self, @@ -441,18 +453,13 @@ def verify( or ``allowed=False`` with ``denial_reason`` and ``jsonrpc_error_code`` on failure. """ - args: Dict[str, Any] = arguments or {} - # PoP bytes cover the wire-args view. Both client and server apply - # strip_none_values to that view so optional arguments with None - # defaults don't crash the Rust canonicalizer and don't silently - # diverge the signed-bytes shape between sides. - pop_args: Dict[str, Any] = strip_none_values(args) + start_ns = time.perf_counter_ns() def _emit_and_return( result: MCPVerificationResult, chain_result: Any = None, - latency_us: int = 0, ) -> MCPVerificationResult: + latency_us = (time.perf_counter_ns() - start_ns) // 1000 if self._control_plane: try: self._control_plane.emit_for_enforcement( @@ -460,8 +467,24 @@ def _emit_and_return( ) except Exception: logger.warning("Control plane emission failed for '%s'; audit event lost", result.tool, exc_info=True) + if self._latency_observer is not None: + try: + self._latency_observer(deepcopy(result), latency_us) + except Exception: + logger.warning( + "Latency observer failed for '%s'", + result.tool, + exc_info=True, + ) return result + args: Dict[str, Any] = arguments or {} + # PoP bytes cover the wire-args view. Both client and server apply + # strip_none_values to that view so optional arguments with None + # defaults don't crash the Rust canonicalizer and don't silently + # diverge the signed-bytes shape between sides. + pop_args: Dict[str, Any] = strip_none_values(args) + # ------------------------------------------------------------------ # Step 1: extract Tenuo envelope from params._meta # ------------------------------------------------------------------ @@ -676,8 +699,6 @@ def _emit_and_return( # ------------------------------------------------------------------ # Step 6: authorize # ------------------------------------------------------------------ - import time - start_ns = time.perf_counter_ns() chain_result = None result: MCPVerificationResult @@ -730,7 +751,6 @@ def _emit_and_return( ), jsonrpc_error_code=-32001, ), - latency_us=(time.perf_counter_ns() - start_ns) // 1000, ) logger.debug("MCP call authorized for '%s' (warrant=%s)", tool_name, warrant_id) @@ -861,8 +881,7 @@ def _emit_and_return( jsonrpc_error_code=-32001, ) - latency_us = (time.perf_counter_ns() - start_ns) // 1000 - return _emit_and_return(result, chain_result=chain_result, latency_us=latency_us) + return _emit_and_return(result, chain_result=chain_result) def verify_or_raise( self, diff --git a/tenuo-python/tests/adapters/test_mcp_server.py b/tenuo-python/tests/adapters/test_mcp_server.py index 5277efe5..1905e69d 100644 --- a/tenuo-python/tests/adapters/test_mcp_server.py +++ b/tenuo-python/tests/adapters/test_mcp_server.py @@ -945,6 +945,116 @@ def test_default_nonce_store_used_when_enabled( assert "replay" in (result2.denial_reason or "").lower() +# --------------------------------------------------------------------------- +# MCPVerifier latency observer +# --------------------------------------------------------------------------- + + +class TestLatencyObserver: + def test_early_denial_observes_end_to_end_latency( + self, authorizer: Authorizer + ) -> None: + from unittest.mock import MagicMock, patch + + observed: list[tuple[MCPVerificationResult, int]] = [] + mock_cp = MagicMock() + verifier = MCPVerifier( + authorizer=authorizer, + control_plane=mock_cp, + latency_observer=lambda result, latency_us: observed.append( + (result, latency_us) + ), + ) + + with patch( + "tenuo.mcp.server.time.perf_counter_ns", + side_effect=[1_000_000, 6_000_000], + ): + result = verifier.verify("read_file", {"path": "/x"}) + + assert not result.allowed + assert observed == [(result, 5_000)] + assert mock_cp.emit_for_enforcement.call_args.kwargs["latency_us"] == 5_000 + + def test_authorized_call_observes_latency( + self, + authorizer: Authorizer, + simple_warrant: Warrant, + agent_key: SigningKey, + ) -> None: + from unittest.mock import MagicMock + + observer = MagicMock() + verifier = MCPVerifier( + authorizer=authorizer, + control_plane=None, + latency_observer=observer, + ) + tool_args = {"path": "/data/f.txt"} + arguments, meta = _make_arguments( + simple_warrant, agent_key, "read_file", tool_args + ) + + result = verifier.verify("read_file", arguments, meta=meta) + + assert result.allowed + observer.assert_called_once() + observed_result, latency_us = observer.call_args.args + assert observed_result == result + assert observed_result is not result + assert latency_us >= 0 + + def test_observer_failure_does_not_change_result_or_audit( + self, + authorizer: Authorizer, + caplog: pytest.LogCaptureFixture, + ) -> None: + from unittest.mock import MagicMock + + def mutate_and_raise(result: MCPVerificationResult, latency_us: int) -> None: + result.allowed = True + result.clean_arguments["path"] = "/mutated" + raise RuntimeError("metrics unavailable") + + observer = MagicMock(side_effect=mutate_and_raise) + mock_cp = MagicMock() + verifier = MCPVerifier( + authorizer=authorizer, + control_plane=mock_cp, + latency_observer=observer, + ) + + result = verifier.verify("read_file", {"path": "/x"}) + + assert not result.allowed + assert result.clean_arguments == {"path": "/x"} + observer.assert_called_once() + mock_cp.emit_for_enforcement.assert_called_once() + assert "Latency observer failed for 'read_file'" in caplog.text + + def test_falsey_callable_observer_is_invoked( + self, authorizer: Authorizer + ) -> None: + class FalseyObserver: + def __init__(self) -> None: + self.calls = 0 + + def __bool__(self) -> bool: + return False + + def __call__( + self, result: MCPVerificationResult, latency_us: int + ) -> None: + self.calls += 1 + + observer = FalseyObserver() + verifier = MCPVerifier(authorizer=authorizer, latency_observer=observer) + + verifier.verify("read_file", {"path": "/x"}) + + assert observer.calls == 1 + + # --------------------------------------------------------------------------- # Control plane emission coverage tests # ---------------------------------------------------------------------------