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

	Facebook
	Twitter
	Linkedin
	E-Mail

Agentic AI > MCP Protocol > MCP Server Audit Logging and Compliance

MCP Server Audit Logging and Compliance

Author: Venkata Sudhakar

Enterprise deployments of MCP servers require full audit trails. Regulators and internal compliance teams need to know who called which tool, with what inputs, and what the outcome was. An MCP server with audit logging writes a structured record for every tool invocation - capturing the agent identity, tool name, arguments, result, and timestamp - and ships these records to a centralised log store.

ShopMax India is subject to GST compliance requirements and internal audit policies. Every tool call on the pricing and orders MCP server must be logged so that investigators can reconstruct what any agent did during a given time window. The example below logs all tool calls to Google Cloud Logging with structured fields, making them queryable in Log Explorer and alertable via Log-based alerts.

The below example shows an MCP server with a reusable audit logging decorator that captures every tool call and writes a structured entry to Cloud Logging before returning the result.

# Install: pip install mcp google-cloud-logging
import asyncio
import functools
import time
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp import types
import google.cloud.logging

# Setup Cloud Logging
log_client = google.cloud.logging.Client(project="shopmax-india")
logger = log_client.logger("mcp-audit")

app = Server("shopmax-audit-mcp")

def audit_log(agent_id: str, tool_name: str, arguments: dict,
               result_summary: str, duration_ms: float, success: bool):
    logger.log_struct({
        "event": "mcp_tool_call",
        "agent_id": agent_id,
        "tool": tool_name,
        "arguments": arguments,
        "result_summary": result_summary,
        "duration_ms": round(duration_ms, 2),
        "success": success
    }, severity="INFO" if success else "WARNING")

# Simulated agent context (in practice, extracted from request headers or cert CN)
AGENT_ID = "shopmax-ops-agent"

@app.list_tools()
async def list_tools():
    return [
        types.Tool(
            name="update_order_status",
            description="Update the status of an order",
            inputSchema={
                "type": "object",
                "properties": {
                    "order_id": {"type": "string"},
                    "status": {"type": "string",
                        "description": "New status: CONFIRMED, SHIPPED, DELIVERED, CANCELLED"}
                },
                "required": ["order_id", "status"]
            }
        ),
        types.Tool(
            name="apply_discount",
            description="Apply a discount to an order",
            inputSchema={
                "type": "object",
                "properties": {
                    "order_id": {"type": "string"},
                    "discount_pct": {"type": "number"}
                },
                "required": ["order_id", "discount_pct"]
            }
        )
    ]

@app.call_tool()
async def call_tool(name: str, arguments: dict):
    start = time.time()
    success = True
    result_summary = ""
    try:
        if name == "update_order_status":
            # Actual update logic here
            result_summary = "Order {} set to {}".format(
                arguments["order_id"], arguments["status"])
            content = [types.TextContent(type="text", text=result_summary)]

elif name == "apply_discount":
            result_summary = "Discount {}% applied to {}".format(
                arguments["discount_pct"], arguments["order_id"])
            content = [types.TextContent(type="text", text=result_summary)]

else:
            result_summary = "Unknown tool: " + name
            content = [types.TextContent(type="text", text=result_summary)]

except Exception as e:
        success = False
        result_summary = "ERROR: " + str(e)
        content = [types.TextContent(type="text", text=result_summary)]

finally:
        duration_ms = (time.time() - start) * 1000
        audit_log(AGENT_ID, name, arguments, result_summary, duration_ms, success)

return content

async def main():
    async with stdio_server() as streams:
        await app.run(streams[0], streams[1], app.create_initialization_options())

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

It gives the following output,

# Agent calls update_order_status:
Tool: update_order_status({order_id: "ORD-5501", status: "SHIPPED"})
Result: Order ORD-5501 set to SHIPPED

# Cloud Logging entry written:
{
  "event": "mcp_tool_call",
  "agent_id": "shopmax-ops-agent",
  "tool": "update_order_status",
  "arguments": {"order_id": "ORD-5501", "status": "SHIPPED"},
  "result_summary": "Order ORD-5501 set to SHIPPED",
  "duration_ms": 12.4,
  "success": true
}

Route MCP audit logs to a dedicated Cloud Logging bucket with a long retention period (1-7 years depending on compliance requirements) and lock the bucket to prevent deletion. Create Log-based metrics to count tool calls per agent and alert on anomalies such as a single agent calling high-risk tools more than N times per hour. Export logs to BigQuery for compliance reports that correlate agent activity with business events such as price changes or order cancellations.

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