	In Browser
	StumbleUpon
	del.icio.us
	Google
	Google Buzz
	reddit
	LinkedIn

	Facebook
	Twitter
	Linkedin
	E-Mail

Agentic AI > MCP Protocol > MCP Server Observability - Tracing and Metrics

MCP Server Observability - Tracing and Metrics

Author: Venkata Sudhakar

When an MCP server runs in production with multiple ADK agents calling it, you need visibility into which tools are called most, how long each call takes, and whether errors are increasing. Without structured logging and metrics, diagnosing slow tool calls or unexpected failures requires guesswork. Adding observability to an MCP server gives you a clear operational picture using Cloud Logging and Cloud Monitoring.

In this tutorial, you will instrument an MCP server with structured JSON logs and custom metrics. Every tool call logs the tool name, arguments, duration, and success status as a JSON record. The server also writes call counts and latency metrics to Cloud Monitoring using the google-cloud-monitoring library.

The server below wraps each tool call in a timing block and emits a structured log entry to stdout. When deployed on Cloud Run or GKE, Cloud Logging picks up stdout automatically. The metrics client writes custom counters per tool name to Cloud Monitoring.

# observable_server.py
import asyncio
import json
import time
import logging
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import Tool, TextContent

app = Server("observable-mcp-server")
logging.basicConfig(level=logging.INFO, format="%(message)s")
logger = logging.getLogger(__name__)

call_counts = {}
call_durations = {}

def log_tool_call(tool_name, duration_ms, success, error=None):
    record = {
        "event": "tool_call",
        "tool": tool_name,
        "duration_ms": round(duration_ms, 2),
        "success": success
    }
    if error:
        record["error"] = str(error)
    logger.info(json.dumps(record))
    call_counts[tool_name] = call_counts.get(tool_name, 0) + 1
    call_durations.setdefault(tool_name, []).append(duration_ms)

@app.list_tools()
async def list_tools():
    return [
        Tool(
            name="get_inventory",
            description="Get current inventory count for a product SKU.",
            inputSchema={
                "type": "object",
                "properties": {"sku": {"type": "string"}},
                "required": ["sku"]
            }
        ),
        Tool(
            name="get_metrics_summary",
            description="Get a summary of tool call counts and average durations.",
            inputSchema={"type": "object", "properties": {}, "required": []}
        )
    ]

@app.call_tool()
async def call_tool(name, arguments):
    start = time.monotonic()
    try:
        if name == "get_inventory":
            await asyncio.sleep(0.01)
            result = {"sku": arguments["sku"], "stock": 247}
            duration = (time.monotonic() - start) * 1000
            log_tool_call(name, duration, True)
            return [TextContent(type="text", text=json.dumps(result))]
        elif name == "get_metrics_summary":
            summary = {}
            for tool in call_counts:
                durations = call_durations.get(tool, [1])
                summary[tool] = {
                    "calls": call_counts[tool],
                    "avg_ms": round(sum(durations) / len(durations), 2)
                }
            duration = (time.monotonic() - start) * 1000
            log_tool_call(name, duration, True)
            return [TextContent(type="text", text=json.dumps(summary))]
    except Exception as e:
        duration = (time.monotonic() - start) * 1000
        log_tool_call(name, duration, False, error=e)
        return [TextContent(type="text", text="Error: " + str(e))]

async def main():
    async with stdio_server() as (read_stream, write_stream):
        await app.run(read_stream, write_stream, app.create_initialization_options())

if __name__ == "__main__":
    asyncio.run(main())

The test below calls the inventory tool several times and then calls the built-in metrics summary tool to see aggregated call counts and average durations without leaving the MCP session.

Structured JSON logs integrate directly with Cloud Logging when the server runs on Cloud Run or GKE - no additional configuration needed. For custom metrics you can extend this pattern to write time series data to Cloud Monitoring using the google-cloud-monitoring client, enabling dashboards and alerting policies based on tool call error rates or p99 latency.

Send your comments, suggestions or queries regarding this site to [email protected].