feat: add signed MCP audit receipts

openbb_platform/extensions/mcp_server/README.md

@@ -148,6 +148,55 @@ custom_auth = BearerProvider(...)
 mcp_server = create_mcp_server(settings, fastapi_app, auth=custom_auth)
 ```
 
+### Signed Audit Receipts
+
+The MCP server can emit signed JSON Lines receipts for tool calls. This is disabled by default. When enabled, each receipt records the tool name, argument hash, timestamp, outcome, duration, principal, public key, and Ed25519 signature.
+
+By default, raw tool arguments are not written to the receipt file. The server stores only `arguments_sha256`, which lets operators verify what was called without accidentally persisting secrets or sensitive query details. Set `OPENBB_MCP_AUDIT_RECEIPTS_INCLUDE_ARGUMENTS=true` only when your deployment is allowed to retain raw arguments.
+
+Generate a base64 raw Ed25519 private key:
+
+```python
+import base64
+
+from cryptography.hazmat.primitives import serialization
+from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
+
+key = Ed25519PrivateKey.generate()
+raw = key.private_bytes(
+    encoding=serialization.Encoding.Raw,
+    format=serialization.PrivateFormat.Raw,
+    encryption_algorithm=serialization.NoEncryption(),
+)
+print(base64.b64encode(raw).decode("ascii"))
+```
+
+Enable receipts:
+
+```env
+OPENBB_MCP_AUDIT_RECEIPTS_ENABLED=true
+OPENBB_MCP_AUDIT_RECEIPTS_PATH="/var/log/openbb/mcp-audit.jsonl"
+OPENBB_MCP_AUDIT_RECEIPTS_PRIVATE_KEY="<base64-ed25519-private-key>"
+OPENBB_MCP_AUDIT_RECEIPTS_PRINCIPAL="research-agent"
+```
+
+Example receipt shape:
+
+```json
+{
+  "receipt_id": "8b1d9b3a-72a4-4f11-9f37-8a13cb55d984",
+  "issued_at": "2026-06-12T00:00:00+00:00",
+  "tool_name": "equity_price_historical",
+  "arguments_sha256": "7e3f...",
+  "status": "success",
+  "policy_decision": "allow",
+  "duration_ms": 21.4,
+  "principal": "research-agent",
+  "public_key": "...",
+  "signature": "..."
+}
+```
+
 ### Advanced Configuration: Lists and Dictionaries
 
 For settings that accept a list or a dictionary, you have two flexible formats for defining them in both command-line arguments and environment variables.
@@ -229,6 +278,11 @@ All settings in the `MCPSettings` model can be configured via the `mcp_settings.
 | `default_skills_dir` | `OPENBB_MCP_DEFAULT_SKILLS_DIR` | string | *(bundled skills dir)* | Path to a directory of bundled skill files. Set to `null` to disable. |
 | `skills_reload` | `OPENBB_MCP_SKILLS_RELOAD` | boolean | `False` | Reload skill files on every read (useful during development). |
 | `skills_providers` | `OPENBB_MCP_SKILLS_PROVIDERS` | list[string] | `None` | Vendor skill provider short-names to load (e.g. `["claude", "cursor"]`). |
+| `audit_receipts_enabled` | `OPENBB_MCP_AUDIT_RECEIPTS_ENABLED` | boolean | `False` | Enable signed JSONL audit receipts for MCP tool calls. |
+| `audit_receipts_path` | `OPENBB_MCP_AUDIT_RECEIPTS_PATH` | string | `None` | Path where audit receipts are appended. Required when receipts are enabled. |
+| `audit_receipts_private_key` | `OPENBB_MCP_AUDIT_RECEIPTS_PRIVATE_KEY` | string | `None` | Ed25519 signing key. Accepts PEM, base64 raw bytes, or hex raw bytes. Required when receipts are enabled. |
+| `audit_receipts_principal` | `OPENBB_MCP_AUDIT_RECEIPTS_PRINCIPAL` | string | `"unknown"` | Principal identifier written to receipts. |
+| `audit_receipts_include_arguments` | `OPENBB_MCP_AUDIT_RECEIPTS_INCLUDE_ARGUMENTS` | boolean | `False` | Include canonical raw arguments in receipts. Defaults to hashing arguments only. |
 | `cache_expiration_seconds` | `OPENBB_MCP_CACHE_EXPIRATION_SECONDS` | float | `None` | Cache expiration time in seconds. `0` to disable. |
 | `on_duplicate_tools` | `OPENBB_MCP_ON_DUPLICATE_TOOLS` | string | `None` | Behavior for duplicate tools (`warn`, `error`, `replace`, `ignore`). |
 | `on_duplicate_resources` | `OPENBB_MCP_ON_DUPLICATE_RESOURCES` | string | `None` | Behavior for duplicate resources. |

openbb_platform/extensions/mcp_server/openbb_mcp_server/app/app.py

@@ -55,6 +55,7 @@
 from openbb_mcp_server.models.tools import CategoryInfo, SubcategoryInfo, ToolInfo
 from openbb_mcp_server.service.mcp_service import MCPService
 from openbb_mcp_server.utils.app_import import parse_args
+from openbb_mcp_server.utils.audit import AuditReceiptMiddleware
 from openbb_mcp_server.utils.fastapi import (
     get_api_prefix,
     process_fastapi_routes_for_mcp,
@@ -533,6 +534,9 @@ def customize_components(
         **fastmcp_kwargs,
     )
 
+    if settings.audit_receipts_enabled:
+        mcp.add_middleware(AuditReceiptMiddleware.from_settings(settings))
+
     # Disable ALL non-admin tools first, then selectively re-enable.
     all_registered = category_index.all_tool_names()
     if all_registered:

openbb_platform/extensions/mcp_server/openbb_mcp_server/models/audit.py

@@ -0,0 +1,27 @@
+"""Audit receipt models for MCP tool calls."""
+
+from typing import Any, Literal
+
+from pydantic import BaseModel, Field
+
+
+class AuditReceipt(BaseModel):
+    """Tamper-evident receipt for a single MCP tool call."""
+
+    receipt_id: str = Field(description="Unique receipt identifier.")
+    issued_at: str = Field(description="UTC timestamp when the receipt was issued.")
+    tool_name: str = Field(description="MCP tool name that was called.")
+    arguments_sha256: str = Field(description="SHA-256 digest of canonical arguments.")
+    status: Literal["success", "error"] = Field(description="Tool call outcome.")
+    policy_decision: str = Field(default="allow", description="Policy decision.")
+    duration_ms: float = Field(description="Tool call duration in milliseconds.")
+    principal: str = Field(description="Agent or user principal identifier.")
+    public_key: str = Field(description="Base64-encoded Ed25519 public key.")
+    signature: str = Field(description="Base64-encoded Ed25519 signature.")
+    session_id: str | None = Field(default=None, description="MCP session id, if known.")
+    request_id: str | None = Field(default=None, description="Request id, if known.")
+    error_type: str | None = Field(default=None, description="Exception type on failure.")
+    arguments: dict[str, Any] | None = Field(
+        default=None,
+        description="Canonical arguments, included only when explicitly configured.",
+    )

openbb_platform/extensions/mcp_server/openbb_mcp_server/models/settings.py

@@ -171,6 +171,33 @@ class MCPSettings(BaseModel):
         alias="OPENBB_MCP_SKILLS_PROVIDERS",
     )
 
+    # Audit receipt configuration
+    audit_receipts_enabled: bool = Field(
+        default=False,
+        description="Enable signed JSONL audit receipts for MCP tool calls.",
+        alias="OPENBB_MCP_AUDIT_RECEIPTS_ENABLED",
+    )
+    audit_receipts_path: str | None = Field(
+        default=None,
+        description="Path to a JSONL file where signed audit receipts are appended.",
+        alias="OPENBB_MCP_AUDIT_RECEIPTS_PATH",
+    )
+    audit_receipts_private_key: str | None = Field(
+        default=None,
+        description="Ed25519 private key for receipt signing. Accepts PEM, base64 raw bytes, or hex raw bytes.",
+        alias="OPENBB_MCP_AUDIT_RECEIPTS_PRIVATE_KEY",
+    )
+    audit_receipts_principal: str = Field(
+        default="unknown",
+        description="Principal identifier recorded in audit receipts.",
+        alias="OPENBB_MCP_AUDIT_RECEIPTS_PRINCIPAL",
+    )
+    audit_receipts_include_arguments: bool = Field(
+        default=False,
+        description="If True, include canonical tool arguments in receipts. Defaults to hashing arguments only.",
+        alias="OPENBB_MCP_AUDIT_RECEIPTS_INCLUDE_ARGUMENTS",
+    )
+
     module_exclusion_map: dict[str, str] | None = Field(
         default=None,
         description="Key:Value pairs mapping API Tags with their Python module names."

openbb_platform/extensions/mcp_server/openbb_mcp_server/utils/audit.py

@@ -0,0 +1,220 @@
+"""Signed audit receipts for MCP tool calls."""
+
+from __future__ import annotations
+
+import base64
+import hashlib
+import json
+import threading
+import time
+import uuid
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Any
+
+from cryptography.exceptions import InvalidSignature
+from cryptography.hazmat.primitives import serialization
+from cryptography.hazmat.primitives.asymmetric.ed25519 import (
+    Ed25519PrivateKey,
+    Ed25519PublicKey,
+)
+from fastmcp.server.middleware import Middleware, MiddlewareContext
+
+from openbb_mcp_server.models.audit import AuditReceipt
+from openbb_mcp_server.models.settings import MCPSettings
+
+
+def _jsonable(value: Any) -> Any:
+    """Return a stable JSON-compatible representation."""
+    if value is None or isinstance(value, str | int | float | bool):
+        return value
+    if isinstance(value, dict):
+        return {
+            str(k): _jsonable(v)
+            for k, v in sorted(value.items(), key=lambda item: str(item[0]))
+        }
+    if isinstance(value, list | tuple | set):
+        return [_jsonable(v) for v in value]
+    if hasattr(value, "model_dump"):
+        return _jsonable(value.model_dump(mode="json"))
+    return repr(value)
+
+
+def canonical_json(data: dict[str, Any]) -> bytes:
+    """Serialize data into canonical JSON bytes."""
+    return json.dumps(
+        _jsonable(data),
+        ensure_ascii=False,
+        separators=(",", ":"),
+        sort_keys=True,
+    ).encode("utf-8")
+
+
+def hash_arguments(arguments: dict[str, Any] | None) -> str:
+    """Return the SHA-256 hash of canonical tool arguments."""
+    return hashlib.sha256(canonical_json(arguments or {})).hexdigest()
+
+
+def _b64encode(data: bytes) -> str:
+    return base64.b64encode(data).decode("ascii")
+
+
+def _b64decode(value: str) -> bytes:
+    return base64.b64decode(value.encode("ascii"), validate=True)
+
+
+def load_ed25519_private_key(value: str) -> Ed25519PrivateKey:
+    """Load an Ed25519 private key from PEM, base64 raw bytes, or hex raw bytes."""
+    value = value.strip()
+    if not value:
+        raise ValueError("Ed25519 private key is required.")
+
+    if value.startswith("-----BEGIN"):
+        key = serialization.load_pem_private_key(value.encode("utf-8"), password=None)
+        if not isinstance(key, Ed25519PrivateKey):
+            raise ValueError("Private key must be an Ed25519 key.")
+        return key
+
+    try:
+        raw = bytes.fromhex(value)
+    except ValueError:
+        raw = _b64decode(value)
+
+    if len(raw) != 32:
+        raise ValueError("Ed25519 raw private key must be exactly 32 bytes.")
+    return Ed25519PrivateKey.from_private_bytes(raw)
+
+
+def public_key_base64(private_key: Ed25519PrivateKey) -> str:
+    """Return the base64-encoded raw public key for an Ed25519 private key."""
+    public_key = private_key.public_key().public_bytes(
+        encoding=serialization.Encoding.Raw,
+        format=serialization.PublicFormat.Raw,
+    )
+    return _b64encode(public_key)
+
+
+def sign_receipt_payload(
+    payload: dict[str, Any], private_key: Ed25519PrivateKey
+) -> AuditReceipt:
+    """Sign a receipt payload and return an AuditReceipt model."""
+    signed_payload = {k: v for k, v in payload.items() if v is not None}
+    signed_payload["public_key"] = public_key_base64(private_key)
+    signature = private_key.sign(canonical_json(signed_payload))
+    return AuditReceipt(**signed_payload, signature=_b64encode(signature))
+
+
+def verify_receipt(receipt: AuditReceipt | dict[str, Any]) -> bool:
+    """Verify an audit receipt signature."""
+    receipt_model = (
+        receipt if isinstance(receipt, AuditReceipt) else AuditReceipt.model_validate(receipt)
+    )
+    payload = receipt_model.model_dump(exclude={"signature"}, exclude_none=True)
+    public_key = Ed25519PublicKey.from_public_bytes(_b64decode(receipt_model.public_key))
+    try:
+        public_key.verify(_b64decode(receipt_model.signature), canonical_json(payload))
+    except InvalidSignature:
+        return False
+    return True
+
+
+class JSONLAuditReceiptWriter:
+    """Append signed audit receipts to a JSON Lines file."""
+
+    def __init__(self, path: str | Path) -> None:
+        self.path = Path(path).expanduser()
+        self._lock = threading.Lock()
+
+    def write(self, receipt: AuditReceipt) -> None:
+        """Write one receipt as a JSONL record."""
+        self.path.parent.mkdir(parents=True, exist_ok=True)
+        line = receipt.model_dump_json(exclude_none=True)
+        with self._lock, self.path.open("a", encoding="utf-8") as f:
+            f.write(line + "\n")
+
+
+def _context_attr(context: MiddlewareContext, name: str) -> str | None:
+    fastmcp_context = getattr(context, "fastmcp_context", None)
+    for source in (context, fastmcp_context):
+        value = getattr(source, name, None)
+        if value:
+            return str(value)
+    return None
+
+
+def _issued_at(context: MiddlewareContext) -> str:
+    timestamp = getattr(context, "timestamp", None)
+    if isinstance(timestamp, datetime):
+        return timestamp.astimezone(timezone.utc).isoformat()
+    return datetime.now(timezone.utc).isoformat()
+
+
+class AuditReceiptMiddleware(Middleware):
+    """FastMCP middleware that emits signed receipts for tool calls."""
+
+    def __init__(
+        self,
+        *,
+        private_key: Ed25519PrivateKey,
+        receipt_path: str | Path,
+        principal: str = "unknown",
+        include_arguments: bool = False,
+    ) -> None:
+        self.private_key = private_key
+        self.writer = JSONLAuditReceiptWriter(receipt_path)
+        self.principal = principal or "unknown"
+        self.include_arguments = include_arguments
+
+    @classmethod
+    def from_settings(cls, settings: MCPSettings) -> AuditReceiptMiddleware:
+        """Create audit middleware from MCP settings."""
+        if not settings.audit_receipts_private_key:
+            raise ValueError(
+                "OPENBB_MCP_AUDIT_RECEIPTS_PRIVATE_KEY is required when "
+                "audit receipts are enabled."
+            )
+        if not settings.audit_receipts_path:
+            raise ValueError(
+                "OPENBB_MCP_AUDIT_RECEIPTS_PATH is required when audit receipts "
+                "are enabled."
+            )
+        return cls(
+            private_key=load_ed25519_private_key(settings.audit_receipts_private_key),
+            receipt_path=settings.audit_receipts_path,
+            principal=settings.audit_receipts_principal,
+            include_arguments=settings.audit_receipts_include_arguments,
+        )
+
+    async def on_call_tool(self, context: MiddlewareContext, call_next: Any) -> Any:
+        """Record a signed receipt after each tool call."""
+        message = getattr(context, "message", None)
+        tool_name = getattr(message, "name", "unknown")
+        arguments = _jsonable(getattr(message, "arguments", {}) or {})
+        started = time.perf_counter()
+        status = "success"
+        error_type = None
+
+        try:
+            return await call_next(context)
+        except Exception as exc:
+            status = "error"
+            error_type = exc.__class__.__name__
+            raise
+        finally:
+            payload: dict[str, Any] = {
+                "receipt_id": str(uuid.uuid4()),
+                "issued_at": _issued_at(context),
+                "tool_name": tool_name,
+                "arguments_sha256": hash_arguments(arguments),
+                "status": status,
+                "policy_decision": "allow",
+                "duration_ms": round((time.perf_counter() - started) * 1000, 3),
+                "principal": self.principal,
+                "session_id": _context_attr(context, "session_id"),
+                "request_id": _context_attr(context, "request_id"),
+                "error_type": error_type,
+            }
+            if self.include_arguments:
+                payload["arguments"] = arguments
+
+            self.writer.write(sign_receipt_payload(payload, self.private_key))

openbb_platform/extensions/mcp_server/pyproject.toml

@@ -18,6 +18,7 @@ openbb-mcp = "openbb_mcp_server.app.app:main"
 python = ">=3.10,<4"
 openbb-core = "^1.6.10"
 fastmcp = ">=3.2.0"
+cryptography = ">=45"
 
 [build-system]
 requires = ["poetry-core>=1.0.0"]