mcp-trace


Namemcp-trace JSON
Version 0.1.7 PyPI version JSON
download
home_pageNone
SummaryOpen-source tracing middleware for MCP servers. Export logs, requests, and tool calls to files, PostgreSQL, Supabase, Contexa, or custom backends with full control over what gets logged.
upload_time2025-11-04 14:33:52
maintainerNone
docs_urlNone
authorNone
requires_python>=3.8
licenseNone
keywords mcp model-context-protocol tracing logging fastmcp middleware observability monitoring postgresql supabase contexa
VCS
bugtrack_url
requirements No requirements were recorded.
Travis-CI No Travis.
coveralls test coverage No coveralls.
            # mcp-trace

<div align="center">
  <img src="images/MCP-TRACE.png" alt="mcp-trace" width="100%"/>
</div>

[![PyPI version](https://badge.fury.io/py/mcp-trace.svg)](https://pypi.org/project/mcp-trace/)
[![License: MIT](https://img.shields.io/badge/License-MIT-yellow.svg)](LICENSE)
[![Python Version](https://img.shields.io/badge/python-3.8%2B-blue.svg)](https://www.python.org/downloads/)

> **Flexible, pluggable tracing middleware for [FastMCP](https://github.com/jlowin/fastmcp) servers.**
> Log every request, tool call, and response to local files, PostgreSQL, Supabase, Contexa, your own backend, or the consoleβ€”with full control over what gets logged, including user identification and PII redaction.

---

## Table of Contents

- [Features](#features)
- [Quickstart](#quickstart)
- [Adapters](#adapters)
  - [Contexa Adapter](#contexa-adapter)
  - [File Adapter](#file-adapter)
  - [Console Adapter](#console-adapter)
  - [PostgreSQL Adapter](#postgresql-adapter)
  - [Supabase Adapter](#supabase-adapter)
  - [Multi-Adapter Example](#multi-adapter-example)
- [Advanced Features](#advanced-features)
  - [User Identification](#user-identification)
  - [PII Redaction](#pii-redaction)
  - [Trace Data Fields](#trace-data-fields)
- [Requirements](#requirements)
- [Contributing](#contributing)
- [License](#license)
- [Links & Acknowledgements](#links--acknowledgements)

---

## Features

- πŸ“¦ **Plug-and-play**: Add tracing to any FastMCP server in seconds
- πŸ—ƒοΈ **Pluggable adapters**: Log to file, PostgreSQL, Supabase, Contexa, console, or your own
- 🧩 **Composable**: Use multiple adapters at once
- πŸ“ **Schema-first**: All traces stored as JSON for easy querying
- πŸ”’ **Privacy-aware**: Built-in PII redaction support
- πŸ‘€ **User identification**: Extract and log user information from requests
- 🌐 **Comprehensive data**: Captures request/response, client info, IP addresses, errors, and more

---

## Quickstart

### Installation

```sh
pip install mcp-trace
```

### Minimal Example (File Adapter)

```python
from mcp.server import FastMCP
from mcp_trace import TraceMiddleware, FileAdapter

mcp = FastMCP("My MCP Server")

trace_adapter = FileAdapter("trace.log")
trace_middleware = TraceMiddleware(adapter=trace_adapter).init(mcp)

@mcp.tool()
def hello(name: str) -> str:
    return f"Hello, {name}!"

if __name__ == "__main__":
    mcp.run(transport="streamable-http")
```

### Console Adapter Example

```python
from mcp.server import FastMCP
from mcp_trace import TraceMiddleware, ConsoleAdapter

mcp = FastMCP("My MCP Server")

trace_adapter = ConsoleAdapter()
trace_middleware = TraceMiddleware(adapter=trace_adapter).init(mcp)

@mcp.tool()
def hello(name: str) -> str:
    return f"Hello, {name}!"

if __name__ == "__main__":
    mcp.run(transport="streamable-http")
```

### Advanced Example (User Identification & PII Redaction)

```python
from mcp.server import FastMCP
from mcp_trace import TraceMiddleware, ConsoleAdapter

def identify_user(context) -> dict:
    """Identify user from context (e.g., from headers, session, etc.)."""
    try:
        request_context = getattr(context, "request_context", None)
        if request_context:
            request = getattr(request_context, "request", None)
            if request:
                headers = getattr(request, "headers", {}) or {}
                headers_lower = {k.lower(): v for k, v in headers.items()}
                
                user_id = headers_lower.get("x-user-id")
                user_name = headers_lower.get("x-user-name")
                user_email = headers_lower.get("x-user-email")
                
                if user_id:
                    return {
                        "user_id": user_id,
                        "user_name": user_name,
                        "user_email": user_email,
                    }
    except Exception:
        pass
    return None

def redact_pii(trace_data: dict) -> dict:
    """Redact PII from trace data before exporting."""
    # Redact user email
    if "user_email" in trace_data and trace_data["user_email"]:
        trace_data["user_email"] = "***REDACTED***"
    
    # Redact sensitive data from request/response
    if "request" in trace_data and isinstance(trace_data["request"], dict):
        if "password" in trace_data["request"]:
            trace_data["request"]["password"] = "***REDACTED***"
        if "api_key" in trace_data["request"]:
            trace_data["request"]["api_key"] = "***REDACTED***"
    
    return trace_data

mcp = FastMCP("My MCP Server")

# Initialize with identify and redact functions
trace_middleware = TraceMiddleware(
    adapter=ConsoleAdapter(),
    identifyUser=identify_user,
    redact=redact_pii
).init(mcp)

@mcp.tool()
def hello(name: str) -> str:
    return f"Hello, {name}!"

if __name__ == "__main__":
    mcp.run(transport="streamable-http")
```

---

## Adapters

### Contexa Adapter

Send traces to [Contexa](https://contexaai.com/) for cloud-based trace storage and analytics.

**Requirements:**

- Contexa API key (`CONTEXA_API_KEY`)
- Contexa Server ID (`CONTEXA_SERVER_ID`)
- [requests](https://pypi.org/project/requests/)

**Usage:**

You can provide your API key and server ID as environment variables or directly as arguments.

```python
from mcp.server import FastMCP
from mcp_trace import TraceMiddleware, ContexaAdapter

mcp = FastMCP("My MCP Server")

# Option 1: Set environment variables
# import os
# os.environ["CONTEXA_API_KEY"] = "your-api-key"
# os.environ["CONTEXA_SERVER_ID"] = "your-server-id"
# contexa_adapter = ContexaAdapter()

# Option 2: Pass directly
contexa_adapter = ContexaAdapter(
    api_key="your-api-key",
    server_id="your-server-id"
)

trace_middleware = TraceMiddleware(adapter=contexa_adapter).init(mcp)

@mcp.tool()
def hello(name: str) -> str:
    return f"Hello, {name}!"

# On shutdown, ensure all events are sent:
# contexa_adapter.flush(timeout=5)
# contexa_adapter.shutdown()

if __name__ == "__main__":
    mcp.run(transport="streamable-http")
```

### File Adapter

Logs each trace as a JSON line to a file.

```python
from mcp_trace import FileAdapter
trace_adapter = FileAdapter("trace.log")
```

### Console Adapter

Prints each trace to the console in a colorized, readable format.

```python
from mcp_trace import ConsoleAdapter
trace_adapter = ConsoleAdapter()
```

### PostgreSQL Adapter

Store traces in a PostgreSQL table for easy querying and analytics.

**Table schema:**

```sql
CREATE TABLE mcp_traces (
    id SERIAL PRIMARY KEY,
    timestamp TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    session_id TEXT NOT NULL,
    trace_data JSONB NOT NULL
);
```

**Usage:**

```python
from mcp.server import FastMCP
from mcp_trace import TraceMiddleware, PostgresAdapter

mcp = FastMCP("My MCP Server")

psql_adapter = PostgresAdapter(dsn="postgresql://user:pass@host:port/dbname")
trace_middleware = TraceMiddleware(adapter=psql_adapter).init(mcp)

@mcp.tool()
def hello(name: str) -> str:
    return f"Hello, {name}!"

if __name__ == "__main__":
    mcp.run(transport="streamable-http")
```

### Supabase Adapter

Log traces to [Supabase](https://supabase.com/) (PostgreSQL as a service).

**Table schema:** (same as above)

**Install:**

```sh
pip install supabase
```

**Usage:**

```python
from mcp.server import FastMCP
from supabase import create_client
from mcp_trace import TraceMiddleware, SupabaseAdapter

mcp = FastMCP("My MCP Server")

supabase = create_client(SUPABASE_URL, SUPABASE_KEY)
supabase_adapter = SupabaseAdapter(supabase)
trace_middleware = TraceMiddleware(adapter=supabase_adapter).init(mcp)

@mcp.tool()
def hello(name: str) -> str:
    return f"Hello, {name}!"

if __name__ == "__main__":
    mcp.run(transport="streamable-http")
```

### Multi-Adapter Example

Send traces to multiple backends at once:

```python
from mcp.server import FastMCP
from mcp_trace import TraceMiddleware, FileAdapter, PostgresAdapter, SupabaseAdapter, ConsoleAdapter
from supabase import create_client

class MultiAdapter:
    def __init__(self, *adapters):
        self.adapters = adapters
    def export(self, trace_data: dict):
        for adapter in self.adapters:
            adapter.export(trace_data)

mcp = FastMCP("My MCP Server")

file_adapter = FileAdapter("trace.log")
psql_adapter = PostgresAdapter(dsn="postgresql://user:pass@host:port/dbname")
supabase = create_client(SUPABASE_URL, SUPABASE_KEY)
supabase_adapter = SupabaseAdapter(supabase)
console_adapter = ConsoleAdapter()

trace_middleware = TraceMiddleware(
    adapter=MultiAdapter(file_adapter, psql_adapter, supabase_adapter, console_adapter)
).init(mcp)

@mcp.tool()
def hello(name: str) -> str:
    return f"Hello, {name}!"

if __name__ == "__main__":
    mcp.run(transport="streamable-http")
```

---

## Advanced Features

### User Identification

The middleware supports identifying users from request context. Pass an `identifyUser` function that extracts user information:

```python
def identify_user(context) -> dict:
    """Extract user info from context. Can be sync or async."""
    # Example: Extract from headers
    try:
        request_context = getattr(context, "request_context", None)
        if request_context:
            request = getattr(request_context, "request", None)
            if request:
                headers = getattr(request, "headers", {}) or {}
                headers_lower = {k.lower(): v for k, v in headers.items()}
                
                user_id = headers_lower.get("x-user-id")
                if user_id:
                    return {
                        "user_id": user_id,
                        "user_name": headers_lower.get("x-user-name"),
                        "user_email": headers_lower.get("x-user-email"),
                    }
    except Exception:
        pass
    return None

trace_middleware = TraceMiddleware(
    adapter=ConsoleAdapter(),
    identifyUser=identify_user
).init(mcp)
```

### PII Redaction

Protect sensitive data by providing a `redact` function that processes trace data before export:

```python
def redact_pii(trace_data: dict) -> dict:
    """Redact PII from trace data before exporting."""
    # Redact user email
    if "user_email" in trace_data:
        trace_data["user_email"] = "***REDACTED***"
    
    # Redact sensitive request fields
    if "request" in trace_data and isinstance(trace_data["request"], dict):
        if "password" in trace_data["request"]:
            trace_data["request"]["password"] = "***REDACTED***"
        if "api_key" in trace_data["request"]:
            trace_data["request"]["api_key"] = "***REDACTED***"
    
    return trace_data

trace_middleware = TraceMiddleware(
    adapter=ConsoleAdapter(),
    redact=redact_pii
).init(mcp)
```

### Trace Data Fields

The middleware captures comprehensive trace data including:

- **Request info**: `type`, `method`, `timestamp`, `duration`, `session_id`
- **User info**: `user_id`, `user_name`, `user_email` (if `identifyUser` is provided)
- **Client info**: `client_id`, `client_name`, `client_version`
- **Request details**: `request` (with query_params, path_params, url, method)
- **Response data**: `response` (structured or text content)
- **Error info**: `is_error`, `error`
- **Network info**: `ip_address` (from X-Forwarded-For or X-Real-IP headers)
- **Entity info**: `entity_name` (tool/resource/prompt name)
- **Metadata**: Custom metadata dictionary

---

## Requirements

- Python 3.8+
- [mcp](https://github.com/modelcontextprotocol/python-sdk) (with CLI support: `mcp[cli]`)
- [psycopg2-binary](https://pypi.org/project/psycopg2-binary/) (for PostgreSQL adapter)
- [supabase](https://github.com/supabase-community/supabase-py) (for Supabase adapter)
- [requests](https://pypi.org/project/requests/) (for Contexa adapter)
- [pydantic](https://pypi.org/project/pydantic/)

---

## Contributing

We love contributions! Please open issues for bugs or feature requests, and submit pull requests for improvements. See [CONTRIBUTING.md](CONTRIBUTING.md) for guidelines.

---

## License

[MIT](LICENSE)

---

## Links & Acknowledgements

- [MCP Python SDK](https://github.com/modelcontextprotocol/python-sdk) β€” Model Context Protocol Python SDK
- [FastMCP](https://github.com/jlowin/fastmcp) β€” FastMCP server framework

            

Raw data

            {
    "_id": null,
    "home_page": null,
    "name": "mcp-trace",
    "maintainer": null,
    "docs_url": null,
    "requires_python": ">=3.8",
    "maintainer_email": "Akshay Galande <mybytecode@gmail.com>",
    "keywords": "mcp, model-context-protocol, tracing, logging, fastmcp, middleware, observability, monitoring, postgresql, supabase, contexa",
    "author": null,
    "author_email": "Akshay Galande <mybytecode@gmail.com>",
    "download_url": "https://files.pythonhosted.org/packages/99/2c/d76cacf035981991b90984e4f51c9af47555f2088150a86746e6990e6da1/mcp_trace-0.1.7.tar.gz",
    "platform": null,
    "description": "# mcp-trace\n\n<div align=\"center\">\n  <img src=\"images/MCP-TRACE.png\" alt=\"mcp-trace\" width=\"100%\"/>\n</div>\n\n[![PyPI version](https://badge.fury.io/py/mcp-trace.svg)](https://pypi.org/project/mcp-trace/)\n[![License: MIT](https://img.shields.io/badge/License-MIT-yellow.svg)](LICENSE)\n[![Python Version](https://img.shields.io/badge/python-3.8%2B-blue.svg)](https://www.python.org/downloads/)\n\n> **Flexible, pluggable tracing middleware for [FastMCP](https://github.com/jlowin/fastmcp) servers.**\n> Log every request, tool call, and response to local files, PostgreSQL, Supabase, Contexa, your own backend, or the console\u2014with full control over what gets logged, including user identification and PII redaction.\n\n---\n\n## Table of Contents\n\n- [Features](#features)\n- [Quickstart](#quickstart)\n- [Adapters](#adapters)\n  - [Contexa Adapter](#contexa-adapter)\n  - [File Adapter](#file-adapter)\n  - [Console Adapter](#console-adapter)\n  - [PostgreSQL Adapter](#postgresql-adapter)\n  - [Supabase Adapter](#supabase-adapter)\n  - [Multi-Adapter Example](#multi-adapter-example)\n- [Advanced Features](#advanced-features)\n  - [User Identification](#user-identification)\n  - [PII Redaction](#pii-redaction)\n  - [Trace Data Fields](#trace-data-fields)\n- [Requirements](#requirements)\n- [Contributing](#contributing)\n- [License](#license)\n- [Links & Acknowledgements](#links--acknowledgements)\n\n---\n\n## Features\n\n- \ud83d\udce6 **Plug-and-play**: Add tracing to any FastMCP server in seconds\n- \ud83d\uddc3\ufe0f **Pluggable adapters**: Log to file, PostgreSQL, Supabase, Contexa, console, or your own\n- \ud83e\udde9 **Composable**: Use multiple adapters at once\n- \ud83d\udcdd **Schema-first**: All traces stored as JSON for easy querying\n- \ud83d\udd12 **Privacy-aware**: Built-in PII redaction support\n- \ud83d\udc64 **User identification**: Extract and log user information from requests\n- \ud83c\udf10 **Comprehensive data**: Captures request/response, client info, IP addresses, errors, and more\n\n---\n\n## Quickstart\n\n### Installation\n\n```sh\npip install mcp-trace\n```\n\n### Minimal Example (File Adapter)\n\n```python\nfrom mcp.server import FastMCP\nfrom mcp_trace import TraceMiddleware, FileAdapter\n\nmcp = FastMCP(\"My MCP Server\")\n\ntrace_adapter = FileAdapter(\"trace.log\")\ntrace_middleware = TraceMiddleware(adapter=trace_adapter).init(mcp)\n\n@mcp.tool()\ndef hello(name: str) -> str:\n    return f\"Hello, {name}!\"\n\nif __name__ == \"__main__\":\n    mcp.run(transport=\"streamable-http\")\n```\n\n### Console Adapter Example\n\n```python\nfrom mcp.server import FastMCP\nfrom mcp_trace import TraceMiddleware, ConsoleAdapter\n\nmcp = FastMCP(\"My MCP Server\")\n\ntrace_adapter = ConsoleAdapter()\ntrace_middleware = TraceMiddleware(adapter=trace_adapter).init(mcp)\n\n@mcp.tool()\ndef hello(name: str) -> str:\n    return f\"Hello, {name}!\"\n\nif __name__ == \"__main__\":\n    mcp.run(transport=\"streamable-http\")\n```\n\n### Advanced Example (User Identification & PII Redaction)\n\n```python\nfrom mcp.server import FastMCP\nfrom mcp_trace import TraceMiddleware, ConsoleAdapter\n\ndef identify_user(context) -> dict:\n    \"\"\"Identify user from context (e.g., from headers, session, etc.).\"\"\"\n    try:\n        request_context = getattr(context, \"request_context\", None)\n        if request_context:\n            request = getattr(request_context, \"request\", None)\n            if request:\n                headers = getattr(request, \"headers\", {}) or {}\n                headers_lower = {k.lower(): v for k, v in headers.items()}\n                \n                user_id = headers_lower.get(\"x-user-id\")\n                user_name = headers_lower.get(\"x-user-name\")\n                user_email = headers_lower.get(\"x-user-email\")\n                \n                if user_id:\n                    return {\n                        \"user_id\": user_id,\n                        \"user_name\": user_name,\n                        \"user_email\": user_email,\n                    }\n    except Exception:\n        pass\n    return None\n\ndef redact_pii(trace_data: dict) -> dict:\n    \"\"\"Redact PII from trace data before exporting.\"\"\"\n    # Redact user email\n    if \"user_email\" in trace_data and trace_data[\"user_email\"]:\n        trace_data[\"user_email\"] = \"***REDACTED***\"\n    \n    # Redact sensitive data from request/response\n    if \"request\" in trace_data and isinstance(trace_data[\"request\"], dict):\n        if \"password\" in trace_data[\"request\"]:\n            trace_data[\"request\"][\"password\"] = \"***REDACTED***\"\n        if \"api_key\" in trace_data[\"request\"]:\n            trace_data[\"request\"][\"api_key\"] = \"***REDACTED***\"\n    \n    return trace_data\n\nmcp = FastMCP(\"My MCP Server\")\n\n# Initialize with identify and redact functions\ntrace_middleware = TraceMiddleware(\n    adapter=ConsoleAdapter(),\n    identifyUser=identify_user,\n    redact=redact_pii\n).init(mcp)\n\n@mcp.tool()\ndef hello(name: str) -> str:\n    return f\"Hello, {name}!\"\n\nif __name__ == \"__main__\":\n    mcp.run(transport=\"streamable-http\")\n```\n\n---\n\n## Adapters\n\n### Contexa Adapter\n\nSend traces to [Contexa](https://contexaai.com/) for cloud-based trace storage and analytics.\n\n**Requirements:**\n\n- Contexa API key (`CONTEXA_API_KEY`)\n- Contexa Server ID (`CONTEXA_SERVER_ID`)\n- [requests](https://pypi.org/project/requests/)\n\n**Usage:**\n\nYou can provide your API key and server ID as environment variables or directly as arguments.\n\n```python\nfrom mcp.server import FastMCP\nfrom mcp_trace import TraceMiddleware, ContexaAdapter\n\nmcp = FastMCP(\"My MCP Server\")\n\n# Option 1: Set environment variables\n# import os\n# os.environ[\"CONTEXA_API_KEY\"] = \"your-api-key\"\n# os.environ[\"CONTEXA_SERVER_ID\"] = \"your-server-id\"\n# contexa_adapter = ContexaAdapter()\n\n# Option 2: Pass directly\ncontexa_adapter = ContexaAdapter(\n    api_key=\"your-api-key\",\n    server_id=\"your-server-id\"\n)\n\ntrace_middleware = TraceMiddleware(adapter=contexa_adapter).init(mcp)\n\n@mcp.tool()\ndef hello(name: str) -> str:\n    return f\"Hello, {name}!\"\n\n# On shutdown, ensure all events are sent:\n# contexa_adapter.flush(timeout=5)\n# contexa_adapter.shutdown()\n\nif __name__ == \"__main__\":\n    mcp.run(transport=\"streamable-http\")\n```\n\n### File Adapter\n\nLogs each trace as a JSON line to a file.\n\n```python\nfrom mcp_trace import FileAdapter\ntrace_adapter = FileAdapter(\"trace.log\")\n```\n\n### Console Adapter\n\nPrints each trace to the console in a colorized, readable format.\n\n```python\nfrom mcp_trace import ConsoleAdapter\ntrace_adapter = ConsoleAdapter()\n```\n\n### PostgreSQL Adapter\n\nStore traces in a PostgreSQL table for easy querying and analytics.\n\n**Table schema:**\n\n```sql\nCREATE TABLE mcp_traces (\n    id SERIAL PRIMARY KEY,\n    timestamp TIMESTAMPTZ NOT NULL DEFAULT NOW(),\n    session_id TEXT NOT NULL,\n    trace_data JSONB NOT NULL\n);\n```\n\n**Usage:**\n\n```python\nfrom mcp.server import FastMCP\nfrom mcp_trace import TraceMiddleware, PostgresAdapter\n\nmcp = FastMCP(\"My MCP Server\")\n\npsql_adapter = PostgresAdapter(dsn=\"postgresql://user:pass@host:port/dbname\")\ntrace_middleware = TraceMiddleware(adapter=psql_adapter).init(mcp)\n\n@mcp.tool()\ndef hello(name: str) -> str:\n    return f\"Hello, {name}!\"\n\nif __name__ == \"__main__\":\n    mcp.run(transport=\"streamable-http\")\n```\n\n### Supabase Adapter\n\nLog traces to [Supabase](https://supabase.com/) (PostgreSQL as a service).\n\n**Table schema:** (same as above)\n\n**Install:**\n\n```sh\npip install supabase\n```\n\n**Usage:**\n\n```python\nfrom mcp.server import FastMCP\nfrom supabase import create_client\nfrom mcp_trace import TraceMiddleware, SupabaseAdapter\n\nmcp = FastMCP(\"My MCP Server\")\n\nsupabase = create_client(SUPABASE_URL, SUPABASE_KEY)\nsupabase_adapter = SupabaseAdapter(supabase)\ntrace_middleware = TraceMiddleware(adapter=supabase_adapter).init(mcp)\n\n@mcp.tool()\ndef hello(name: str) -> str:\n    return f\"Hello, {name}!\"\n\nif __name__ == \"__main__\":\n    mcp.run(transport=\"streamable-http\")\n```\n\n### Multi-Adapter Example\n\nSend traces to multiple backends at once:\n\n```python\nfrom mcp.server import FastMCP\nfrom mcp_trace import TraceMiddleware, FileAdapter, PostgresAdapter, SupabaseAdapter, ConsoleAdapter\nfrom supabase import create_client\n\nclass MultiAdapter:\n    def __init__(self, *adapters):\n        self.adapters = adapters\n    def export(self, trace_data: dict):\n        for adapter in self.adapters:\n            adapter.export(trace_data)\n\nmcp = FastMCP(\"My MCP Server\")\n\nfile_adapter = FileAdapter(\"trace.log\")\npsql_adapter = PostgresAdapter(dsn=\"postgresql://user:pass@host:port/dbname\")\nsupabase = create_client(SUPABASE_URL, SUPABASE_KEY)\nsupabase_adapter = SupabaseAdapter(supabase)\nconsole_adapter = ConsoleAdapter()\n\ntrace_middleware = TraceMiddleware(\n    adapter=MultiAdapter(file_adapter, psql_adapter, supabase_adapter, console_adapter)\n).init(mcp)\n\n@mcp.tool()\ndef hello(name: str) -> str:\n    return f\"Hello, {name}!\"\n\nif __name__ == \"__main__\":\n    mcp.run(transport=\"streamable-http\")\n```\n\n---\n\n## Advanced Features\n\n### User Identification\n\nThe middleware supports identifying users from request context. Pass an `identifyUser` function that extracts user information:\n\n```python\ndef identify_user(context) -> dict:\n    \"\"\"Extract user info from context. Can be sync or async.\"\"\"\n    # Example: Extract from headers\n    try:\n        request_context = getattr(context, \"request_context\", None)\n        if request_context:\n            request = getattr(request_context, \"request\", None)\n            if request:\n                headers = getattr(request, \"headers\", {}) or {}\n                headers_lower = {k.lower(): v for k, v in headers.items()}\n                \n                user_id = headers_lower.get(\"x-user-id\")\n                if user_id:\n                    return {\n                        \"user_id\": user_id,\n                        \"user_name\": headers_lower.get(\"x-user-name\"),\n                        \"user_email\": headers_lower.get(\"x-user-email\"),\n                    }\n    except Exception:\n        pass\n    return None\n\ntrace_middleware = TraceMiddleware(\n    adapter=ConsoleAdapter(),\n    identifyUser=identify_user\n).init(mcp)\n```\n\n### PII Redaction\n\nProtect sensitive data by providing a `redact` function that processes trace data before export:\n\n```python\ndef redact_pii(trace_data: dict) -> dict:\n    \"\"\"Redact PII from trace data before exporting.\"\"\"\n    # Redact user email\n    if \"user_email\" in trace_data:\n        trace_data[\"user_email\"] = \"***REDACTED***\"\n    \n    # Redact sensitive request fields\n    if \"request\" in trace_data and isinstance(trace_data[\"request\"], dict):\n        if \"password\" in trace_data[\"request\"]:\n            trace_data[\"request\"][\"password\"] = \"***REDACTED***\"\n        if \"api_key\" in trace_data[\"request\"]:\n            trace_data[\"request\"][\"api_key\"] = \"***REDACTED***\"\n    \n    return trace_data\n\ntrace_middleware = TraceMiddleware(\n    adapter=ConsoleAdapter(),\n    redact=redact_pii\n).init(mcp)\n```\n\n### Trace Data Fields\n\nThe middleware captures comprehensive trace data including:\n\n- **Request info**: `type`, `method`, `timestamp`, `duration`, `session_id`\n- **User info**: `user_id`, `user_name`, `user_email` (if `identifyUser` is provided)\n- **Client info**: `client_id`, `client_name`, `client_version`\n- **Request details**: `request` (with query_params, path_params, url, method)\n- **Response data**: `response` (structured or text content)\n- **Error info**: `is_error`, `error`\n- **Network info**: `ip_address` (from X-Forwarded-For or X-Real-IP headers)\n- **Entity info**: `entity_name` (tool/resource/prompt name)\n- **Metadata**: Custom metadata dictionary\n\n---\n\n## Requirements\n\n- Python 3.8+\n- [mcp](https://github.com/modelcontextprotocol/python-sdk) (with CLI support: `mcp[cli]`)\n- [psycopg2-binary](https://pypi.org/project/psycopg2-binary/) (for PostgreSQL adapter)\n- [supabase](https://github.com/supabase-community/supabase-py) (for Supabase adapter)\n- [requests](https://pypi.org/project/requests/) (for Contexa adapter)\n- [pydantic](https://pypi.org/project/pydantic/)\n\n---\n\n## Contributing\n\nWe love contributions! Please open issues for bugs or feature requests, and submit pull requests for improvements. See [CONTRIBUTING.md](CONTRIBUTING.md) for guidelines.\n\n---\n\n## License\n\n[MIT](LICENSE)\n\n---\n\n## Links & Acknowledgements\n\n- [MCP Python SDK](https://github.com/modelcontextprotocol/python-sdk) \u2014 Model Context Protocol Python SDK\n- [FastMCP](https://github.com/jlowin/fastmcp) \u2014 FastMCP server framework\n",
    "bugtrack_url": null,
    "license": null,
    "summary": "Open-source tracing middleware for MCP servers. Export logs, requests, and tool calls to files, PostgreSQL, Supabase, Contexa, or custom backends with full control over what gets logged.",
    "version": "0.1.7",
    "project_urls": {
        "Bug Tracker": "https://github.com/contexaAI/mcp-trace/issues",
        "Changelog": "https://github.com/contexaAI/mcp-trace/releases",
        "Discussions": "https://github.com/contexaAI/mcp-trace/discussions",
        "Documentation": "https://github.com/contexaAI/mcp-trace#readme",
        "Funding": "https://github.com/sponsors/contexaAI",
        "Homepage": "https://github.com/contexaAI/mcp-trace",
        "Issues": "https://github.com/contexaAI/mcp-trace/issues",
        "Repository": "https://github.com/contexaAI/mcp-trace.git",
        "Source Code": "https://github.com/contexaAI/mcp-trace",
        "Twitter": "https://twitter.com/contexaAI"
    },
    "split_keywords": [
        "mcp",
        " model-context-protocol",
        " tracing",
        " logging",
        " fastmcp",
        " middleware",
        " observability",
        " monitoring",
        " postgresql",
        " supabase",
        " contexa"
    ],
    "urls": [
        {
            "comment_text": null,
            "digests": {
                "blake2b_256": "37d466df0121c4c6da099c786658cacb17adfb2aa7393279f2772cf2b67ab447",
                "md5": "ca0e01aeba0a0bfae10f65b1aaefba19",
                "sha256": "61fb5e05f93593d5a0c38af527d1a0c0c57d6436d298f037af9dc05daaf935ec"
            },
            "downloads": -1,
            "filename": "mcp_trace-0.1.7-py3-none-any.whl",
            "has_sig": false,
            "md5_digest": "ca0e01aeba0a0bfae10f65b1aaefba19",
            "packagetype": "bdist_wheel",
            "python_version": "py3",
            "requires_python": ">=3.8",
            "size": 15437,
            "upload_time": "2025-11-04T14:33:51",
            "upload_time_iso_8601": "2025-11-04T14:33:51.523132Z",
            "url": "https://files.pythonhosted.org/packages/37/d4/66df0121c4c6da099c786658cacb17adfb2aa7393279f2772cf2b67ab447/mcp_trace-0.1.7-py3-none-any.whl",
            "yanked": false,
            "yanked_reason": null
        },
        {
            "comment_text": null,
            "digests": {
                "blake2b_256": "992cd76cacf035981991b90984e4f51c9af47555f2088150a86746e6990e6da1",
                "md5": "88d44ab11970b11c16d23fbabc16af62",
                "sha256": "a34917998f9c1cb6bd7b1d60f3c60b4a605fa0bb66ca30a3a4417c6f845d1689"
            },
            "downloads": -1,
            "filename": "mcp_trace-0.1.7.tar.gz",
            "has_sig": false,
            "md5_digest": "88d44ab11970b11c16d23fbabc16af62",
            "packagetype": "sdist",
            "python_version": "source",
            "requires_python": ">=3.8",
            "size": 17049,
            "upload_time": "2025-11-04T14:33:52",
            "upload_time_iso_8601": "2025-11-04T14:33:52.855708Z",
            "url": "https://files.pythonhosted.org/packages/99/2c/d76cacf035981991b90984e4f51c9af47555f2088150a86746e6990e6da1/mcp_trace-0.1.7.tar.gz",
            "yanked": false,
            "yanked_reason": null
        }
    ],
    "upload_time": "2025-11-04 14:33:52",
    "github": true,
    "gitlab": false,
    "bitbucket": false,
    "codeberg": false,
    "github_user": "contexaAI",
    "github_project": "mcp-trace",
    "travis_ci": false,
    "coveralls": false,
    "github_actions": true,
    "lcname": "mcp-trace"
}
        
Elapsed time: 3.17835s