MCP Auditing

This is a cloud feature — @torknetwork/sdk

MCP tool-call auditing is evaluated by Tork, so each audited call is recorded as capture_mode=cloud, attested_by=tork. There is no dedicated SDK class for MCP Auditing in @torknetwork/sdk or in any published Python package — call the API directly with a Bearer token, in either language, as shown below.

const TORK_API_KEY = process.env.TORK_API_KEY;

// Wrap your MCP tool calls
async function readFile(path) {
  const start = Date.now();
  let result, success = true, errorMessage;

  try {
    result = await mcpServer.callTool('read_file', { path });
    return result;
  } catch (err) {
    success = false;
    errorMessage = err.message;
    throw err;
  } finally {
    await fetch('https://tork.network/api/v1/mcp/audit', {
      method: 'POST',
      headers: {
        Authorization: `Bearer ${TORK_API_KEY}`,
        'Content-Type': 'application/json',
      },
      body: JSON.stringify({
        action: 'log',
        agentId: 'my-agent',
        toolName: 'read_file',
        toolParameters: { path },
        result,
        executionTimeMs: Date.now() - start,
        success,
        errorMessage,
      }),
    });
  }
}

import os
import time
import requests

TORK_API_KEY = os.environ["TORK_API_KEY"]

async def read_file(path: str) -> str:
    start = time.monotonic()
    success = True
    error_message = None
    result = None

    try:
        result = await mcp_server.call_tool("read_file", {"path": path})
        return result
    except Exception as exc:
        success = False
        error_message = str(exc)
        raise
    finally:
        requests.post(
            "https://tork.network/api/v1/mcp/audit",
            headers={"Authorization": f"Bearer {TORK_API_KEY}"},
            json={
                "action": "log",
                "agentId": "my-agent",
                "toolName": "read_file",
                "toolParameters": {"path": path},
                "result": result,
                "executionTimeMs": int((time.monotonic() - start) * 1000),
                "success": success,
                "errorMessage": error_message,
            },
            timeout=10,
        )

# mcp-policies.yaml
policies:
  - name: restrict-filesystem-access
    description: Limit file operations to specific directories
    tool_pattern: "filesystem:*"
    action: BLOCK
    conditions:
      - type: path_not_in
        allowed_paths:
          - "/app/data"
          - "/tmp"
    message: "File access outside allowed directories"

  - name: require-approval-for-writes
    description: Require human approval for write operations
    tool_pattern: "filesystem:write_file"
    action: APPROVAL
    approvers: ["admin@company.com"]

  - name: rate-limit-api-calls
    description: Limit external API calls
    tool_pattern: "http:*"
    action: BLOCK
    conditions:
      - type: rate_exceeded
        limit: 100
        window: "1m"

# Query audit logs
logs = client.mcp.get_audit_logs(
    server_name="filesystem-server",
    tool_name="read_file",
    start_time="2024-01-01T00:00:00Z",
    end_time="2024-01-31T23:59:59Z",
    status=["success", "blocked"]
)

for log in logs:
    print(f"{log.timestamp}: {log.tool_name}")
    print(f"  Agent: {log.agent_id}")
    print(f"  Status: {log.status}")
    print(f"  Duration: {log.duration_ms}ms")
    if log.policy_violations:
        print(f"  Violations: {log.policy_violations}")

# Get MCP metrics
metrics = client.mcp.get_metrics(
    server_name="filesystem-server",
    window="24h"
)

print(f"Total invocations: {metrics.total_invocations}")
print(f"Success rate: {metrics.success_rate}%")
print(f"Average latency: {metrics.avg_latency_ms}ms")
print(f"Policy blocks: {metrics.policy_blocks}")

# Tool breakdown
for tool in metrics.by_tool:
    print(f"  {tool.name}: {tool.invocations} calls")