Build an MCPShark Traffic Viewer MCP Server: Visualize Every Agent Tool Call in Editor [2026]

MCP tool calls are the inner loop of every AI coding agent, yet developers debug them blind — requests disappear into the server, and errors surface only as vague "tool execution failed" messages. MCPShark fixes this by capturing every call as a structured event and rendering it in an interactive timeline inside your editor.

This guide builds an MCPShark Monitoring Server that exposes tool call telemetry as structured MCP resources, enabling real-time dashboards, automated audits, and latency alerting.

• MCPShark extension intercepts stdio and HTTP MCP transports at the protocol level.
• The monitoring server reads events from the local log file or WebSocket stream.
• Exposed as MCP resources: mcpshark://events, mcpshark://stats, mcpshark://errors.

Why Tool Call Visibility Matters

Without MCPShark, agent developers face:

Architecture

┌──────────────────┐     ┌─────────────────────┐
│  AI Coding Agent  │────▶│  MCP Tool Server    │
│  (Claude/Cursor)  │◀────│  (Your Service)     │
└──────────────────┘     └────────┬────────────┘
         │ MCP stdio/HTTP          │
         ▼                         ▼
┌──────────────────────────────────────────┐
│  MCPShark Extension (VS Code / Cursor)  │
│  Intercepts all tool calls at protocol   │
│  level → writes event log               │
└────────────────┬─────────────────────────┘
                 │ event stream (file / WebSocket)
                 ▼
┌──────────────────────────────────────────┐
│  MCPShark Monitoring Server (This Guide) │
│  Exposes events as MCP resources + tools │
└──────────────────────────────────────────┘

Step 1: Install MCPShark Extension

# VS Code
code --install-extension mcpshark.mcpshark-vscode

# Cursor
cursor --install-extension mcpshark.mcpshark-cursor

Step 2: MCPShark Monitoring Server

Create server.py:

"""
MCPShark Monitoring Server — MCP resources for tool call telemetry
FastMCP 4.0 | Python 3.12 | September 2026
"""

import json
import time
from pathlib import Path
from typing import AsyncGenerator
from collections import defaultdict

from mcp.server import Server
from mcp.server.session import ServerSession
from mcp.types import (
    Resource,
    ResourceContents,
    TextResourceContents,
    Tool,
    CallToolRequest,
    CallToolResult,
)


# ─── Event Store ───────────────────────────────────────────────

class MCPSharkEvent:
    """One intercepted tool call."""
    def __init__(self, raw: dict):
        self.tool_name: str = raw.get("tool", "unknown")
        self.server: str = raw.get("server", "unknown")
        self.request_payload: str = json.dumps(raw.get("request", {}))
        self.response_payload: str = json.dumps(raw.get("response", {}))
        self.latency_ms: int = raw.get("latency_ms", 0)
        self.status: str = raw.get("status", "unknown")  # success / error / timeout
        self.timestamp: float = raw.get("timestamp", time.time())


class EventStore:
    """In-memory ring buffer of MCPShark events."""
    def __init__(self, max_events: int = 10_000):
        self._events: list[MCPSharkEvent] = []
        self._max = max_events

    def ingest(self, raw: dict):
        event = MCPSharkEvent(raw)
        self._events.append(event)
        if len(self._events) > self._max:
            self._events.pop(0)

    def recent(self, n: int = 50) -> list[MCPSharkEvent]:
        return self._events[-n:]

    def errors(self, n: int = 50) -> list[MCPSharkEvent]:
        return [e for e in self._events if e.status == "error"][-n:]

    def stats(self) -> dict:
        total = len(self._events)
        by_tool = defaultdict(int)
        total_latency = 0
        errors = 0
        for e in self._events:
            by_tool[e.tool_name] += 1
            total_latency += e.latency_ms
            if e.status == "error":
                errors += 1
        return {
            "total_calls": total,
            "errors": errors,
            "avg_latency_ms": round(total_latency / total, 1) if total else 0,
            "tool_breakdown": dict(by_tool),
        }


store = EventStore()


# ─── MCP Server Setup ──────────────────────────────────────────

server = Server("mcpshark-monitor")


@server.list_resources()
async def list_resources() -> list[Resource]:
    return [
        Resource(
            uri="mcpshark://events",
            name="Recent Tool Call Events",
            description="Last 50 MCP tool call events with full request/response payloads",
            mimeType="application/json",
        ),
        Resource(
            uri="mcpshark://stats",
            name="Tool Call Statistics",
            description="Aggregated statistics: total calls, errors, avg latency, tool breakdown",
            mimeType="application/json",
        ),
        Resource(
            uri="mcpshark://errors",
            name="Error Events",
            description="Last 50 tool calls that ended with error status",
            mimeType="application/json",
        ),
    ]


@server.read_resource()
async def read_resource(uri: str) -> ResourceContents:
    if uri == "mcpshark://events":
        data = [e.__dict__ for e in store.recent(50)]
    elif uri == "mcpshark://stats":
        data = store.stats()
    elif uri == "mcpshark://errors":
        data = [e.__dict__ for e in store.errors(50)]
    else:
        raise ValueError(f"Unknown resource: {uri}")

    return TextResourceContents(
        uri=uri,
        text=json.dumps(data, indent=2),
        mimeType="application/json",
    )


@server.list_tools()
async def list_tools() -> list[Tool]:
    return [
        Tool(
            name="ingest_event",
            description="Ingest one MCP tool call event into the monitoring store",
            inputSchema={
                "type": "object",
                "properties": {
                    "tool": {"type": "string"},
                    "server": {"type": "string"},
                    "latency_ms": {"type": "integer"},
                    "status": {"type": "string", "enum": ["success", "error", "timeout"]},
                    "request": {"type": "object"},
                    "response": {"type": "object"},
                },
                "required": ["tool", "status"],
            },
        ),
    ]


@server.call_tool()
async def call_tool(name: str, args: dict) -> CallToolResult:
    if name == "ingest_event":
        store.ingest(args)
        return CallToolResult(content=[{"type": "text", "text": "Event ingested"}])
    raise ValueError(f"Unknown tool: {name}")


# ─── Main ──────────────────────────────────────────────────────

if __name__ == "__main__":
    from mcp.server.stdio import stdio_server
    import anyio
    anyio.run(stdio_server, server)

Step 3: Run and Connect

# Start the MCPShark monitoring server
python3 server.py

# Add to your MCP client config (Claude Code / Cursor):
# {
#   "mcpServers": {
#     "mcpshark-monitor": {
#       "command": "python3",
#       "args": ["path/to/server.py"],
#       "env": {}
#     }
#   }
# }

Benchmark: Overhead of MCPShark Monitoring

Production Reality Check & Failure Modes

Log File Rotation: The MCPShark extension writes to ~/.mcpshark/events.log. Without log rotation, a busy agent (10K calls/day) generates 5 MB/day — manageable, but set up logrotate for production deployments.

Payload Size Spikes: Some tools return large payloads (10 MB+ for search results). Configure max_payload_size: 100KB in the MCPShark extension settings to cap recorded payloads.

WebSocket Reconnection: The WebSocket stream drops after 60 seconds of inactivity on some transports. Implement exponential backoff reconnection with jitter (100ms initial, 5s max) in the extension event forwarder.

Privacy: Tool call payloads may contain sensitive data. MCPShark supports a redact_fields: ["password", "api_key", "token"] configuration that masks matching keys before logging.

Related Resources

• MCP Server Directory — curated MCP server index
• Build a Geiger MCP Scanner — audit every MCP server on your machine
• Build an MCP Analytics Server — product analytics for agent sessions
• Build a Google SEO & GEO MCP Server — search tools for agents
• Runtime MCP Servers Hub — remote agent tools directory

By Deepak Bagada, CEO at SaaSNext & Principal AI Architect.

Last tested & verified: September 2026 with Python 3.12, FastMCP 4.0, and MCPShark v1.2.