MCP工具调用观测平台

MCP Observatory

MCP Observatory now includes a two-phase execution pattern for high-risk MCP tool calls, with a generic proposer/verifier wrapper that can be reused by any tool:

1. PROPOSE: plan/simulate, evaluate uncertainty/integrity, no side effects.
2. COMMIT: execute side effects only when a signed commit token is valid.

Two-Phase Sequence (Text Diagram)

Client
  -> transfer_funds_propose(amount,to)
      -> scoring(output_instability, numeric_variance, prompt_drift)
      -> decision:
          - blocked: deterministic fallback (create_draft), no side effects
          - allowed: issue signed commit_token bound to tool args hash
  <- {proposal_id, commit_token?}

Client
  -> transfer_funds_commit(proposal_id, commit_token, amount, to)
      -> verify signature + expiry + proposal existence + args_hash binding + nonce replay
      -> if valid: perform side effect (funds transfer)
      -> else: block with explicit reason
  <- commit outcome

New Modules

• mcp_observatory/proposal_commit/hashing.py
  • canonical JSON hashing for stable tool_args_hash
  • normalized prompt_hash
• mcp_observatory/proposal_commit/scoring.py
  • output_instability = 1 - jaccard_similarity
  • numeric_variance from extracted numbers
  • prompt_drift from prompt hash vs baseline
  • weighted renormalized composite_score
  • demo model_generate(prompt, temperature) stub
• mcp_observatory/proposal_commit/token.py
  • HMAC-SHA256 token issue/verify
  • payload fields: token_id, proposal_id, tool_name, tool_args_hash, issued_at, expires_at, nonce, composite_score
• mcp_observatory/proposal_commit/proposer.py
  • generic ToolProposer.propose(...) for any tool name/args
  • deterministic blocked fallback
• mcp_observatory/proposal_commit/verifier.py
  • commit verification and nonce replay protection
• mcp_observatory/proposal_commit/storage.py
  • in-memory storage fallback
  • optional Postgres storage via asyncpg
• mcp_observatory/core/wrapper_api.py
  • generic InvocationWrapperAPI for wrapping either agent or model invocations
  • captures input/output hashes, token/cost metrics, and emits allow/review/block decisions
• mcp_observatory/instrument.py
  • adds instrument_wrapper_api(...) helper for fast wrapper setup
• mcp_observatory/demo/server.py
  • MCP-like tools:
    • transfer_funds_propose
    • transfer_funds_commit
• mcp_observatory/demo/run_demo.py
  • propose -> commit -> replay-attempt demo
• sql/schema.sql
  • Postgres tables: proposals, commits, nonces, tool_prompt_baselines

Security / Verification Rules

Commit verifies all of the following:

• token signature is valid (bad_signature on failure)
• token not expired (expired)
• proposal exists and was allowed (unknown_proposal)
• commit args hash equals token payload args hash (args_hash_mismatch)
• nonce has not already been used (nonce_replay)

Deterministic Fallback on Proposal Block

Blocked proposal response is deterministic and side-effect free:

{
  "status": "blocked",
  "action": "create_draft",
  "reason": "low_integrity",
  "draft": {"tool": "transfer_funds", "amount": 100, "to": "acct_123"}
}

Environment Variables

Neither secret variable has a default: constructing TokenIssuer, TokenVerifier, or CommitTokenManager without an explicit secret and without one of these set raises InsecureDefaultSecretError (mcp_observatory.utils.secrets) rather than silently signing with a hardcoded value.

Running the Demo

Without Postgres (default)

No env vars needed; in-memory store is used.

python -m mcp_observatory.demo.run_demo

With Postgres

1. Set DSN:

export MCP_OBSERVATORY_PG_DSN='postgresql://user:pass@localhost:5432/postgres'

2. Apply schema:

psql "$MCP_OBSERVATORY_PG_DSN" -f sql/schema.sql

3. Run demo:

python -m mcp_observatory.demo.run_demo

Testing

PYTHONPATH=. pytest -q

The suite includes tests for token verification, hash stability, replay protection, and expired-token rejection.

Real-World MCP Scenario Demo (10 End-to-End Flows)

A realistic MCP server example is available at:

• mcp_observatory/demo/real_world_server.py
• executable shim: examples/real_world_mcp_server.py
• executable client: examples/real_world_mcp_client.py
• prompt-to-invocation MVP: examples/prompt_to_mcp_invocation_mvp.py
• OpenAI GPT utility: examples/openai_prompt_to_mcp_invocation.py

It includes:

• 10 distinct prompts mapped to 10 different MCP tool handlers
• per-invocation annotations (e.g. destructiveHint, idempotentHint, openWorldHint)
• proposal/commit execution for HIGH-risk tools (no secondary-response gating)
• irreversible actions never pass a secondary LLM response
• simulated LLM responses and grounding summaries for standard-risk tools
• deterministic fallback routing for blocked/review-required scenarios
• prompt-to-invocation MVP now extracts required tool parameters from user prompts before server invocation
• openai-gpt utility for service selection + parameter extraction + MCP invocation

Run server demo:

python examples/real_world_mcp_server.py

Run client demo (client interacting with server):

python examples/real_world_mcp_client.py

Run prompt -> LLM planner -> server invocation MVP:

python examples/prompt_to_mcp_invocation_mvp.py

Run OpenAI GPT utility (manual, requires OPENAI_API_KEY):

python examples/openai_prompt_to_mcp_invocation.py

Optional (recommended): install OpenAI SDK for first-class client support (client auto-detects and uses SDK when available; otherwise it uses HTTPS fallback):

pip install -e .[openai]

Wrapper API (Agent or Model Invocation)

Use the wrapper to route either agent-side orchestration calls or direct model calls through a single observability envelope.

from mcp_observatory.instrument import instrument_wrapper_api

wrapper = instrument_wrapper_api("my-service")

result = await wrapper.invoke(
    source="agent",
    model="gpt-4o-mini",
    prompt="Generate deployment plan",
    input_payload={"request_id": "abc123", "task": "deployment_plan"},
    call=lambda: {"plan": "blue-green rollout"},
)

print(result.decision.action)  # allow/review/block
print(result.span.cost_usd)

Dual-run measurement is supported with dual_invoke=True and shadow parameters (shadow_source, shadow_model, shadow_agent_params, shadow_model_params, and shadow_call) to compare alternate execution paths and capture disagreement metrics.

The wrapper output (WrapperResult) includes:

• output: raw callable output
• span: captured telemetry metrics (tokens, cost, hashes, timing)
• decision: policy decision suitable for downstream execution routing
• shadow_output and shadow_span (when dual_invoke=True) with comparison metrics on primary span (shadow_disagreement_score, shadow_numeric_variance)

AWS Integration (mcp_observatory.aws)

Common wiring extracted from services that deploy this library on AWS Lambda. Install the extra to use it:

pip install mcp-observatory[aws]

Importing mcp_observatory.aws never requires boto3 — it is imported lazily on first use, so the rest of the library still imports fine without it.

from mcp_observatory.aws import DynamoDBSpanExporter, build_gate

# Writes every populated TraceContext field to a DynamoDB table (name from
# OBSERVATORY_METRICS_TABLE, or pass table_name=) with a TTL attribute.
exporter = DynamoDBSpanExporter(ttl_seconds=90 * 24 * 60 * 60)

# Wires an InMemoryStorage- (or Postgres-, when MCP_OBSERVATORY_PG_DSN is
# set) backed ToolProposer + CommitVerifier sharing one CommitTokenManager.
# secret_env/block_threshold_env let each service keep its own env var
# names; the secret still fails closed per the Environment Variables table
# above.
proposer, verifier, token_manager = build_gate(
    secret_env="OBSERVATORY_SECRET_KEY",
    block_threshold_env="OBSERVATORY_BLOCK_THRESHOLD",
)

There is no observe_model_request-style helper here: the services this was extracted from do not agree on its shape (see CHANGELOG.md 0.3.0), so none is provided.