AgentTrace 🔍
MCP server for tracing, logging, and debugging multi-agent systems — one session, full visibility.
---
What is AgentTrace?
When something breaks in a multi-agent system, finding the root cause is painful. Logs are scattered across agents, tool calls are invisible, and there's no single place to see what actually happened.
AgentTrace solves this by treating a session as the source of truth. Every human input, agent-to-agent call, tool invocation, and human-in-the-loop event is captured under a single session — giving you a complete, queryable trace tree of everything that happened.
Session abc-123
├── [human_input] span-1 "Summarize this report"
│ └── [agent_call] span-2 Orchestrator → ResearchAgent
│ ├── [tool_call] span-3 ResearchAgent → web_search
│ └── [tool_call] span-4 ResearchAgent → read_file
│ └── EVENT: error — "file not found" ← root cause
└── [hitl] span-5 pending approval---
🔒 Need Advanced Authorization?
As an MCP server, AgentTrace acts as the central hub for your multi-agent architecture. If your setup requires strict access controls, granular agent permissions, or specialized authorization flows, check out the
feature/Auth-Enablebranch for our fully authenticated implementation!
---
Features
- 📌 Session-scoped tracing — every run is one session, all events hang off it
- 🌲 Full trace tree reconstruction — visualize the exact call graph via
parent_span_id - 🔎 Cross-session querying — query by agent, error type, span type, or time range across all sessions
- 🔁 Soft session replay — re-inject original inputs into a fresh run for root cause analysis, optionally starting from mid-trace
- 🧩 Protocol-native — built as an MCP server, any MCP-compatible agent gets tracing for free just by connecting
- 🏗️ Production-grade — async SQLAlchemy, PostgreSQL, strict three-layer architecture
---
Tech Stack
| Layer | Technology | | --------------- | ---------------------------- | | MCP Framework | FastMCP 3.x | | Database | PostgreSQL 16 | | ORM | SQLAlchemy (async) | | Migrations | Alembic | | Validation | Pydantic + pydantic-settings | | Package Manager | UV |
---
Setup
Option 1 — Run Locally
Prerequisites: Python 3.11+, PostgreSQL, UV
# Clone the repo
git clone https://github.com/your-username/agenttrace.git
cd agenttrace
# Install dependencies
uv sync
# Copy and configure env
cp .env.example .envEdit .env:
APP_ENV=development
DEBUG=false
HOST=0.0.0.0
PORT=8000
DATABASE_URL=postgresql+asyncpg://postgres:password@localhost:5432/agenttraceRun migrations and start the server:
alembic upgrade head
python -m mcp_tracer.serverServer runs at http://localhost:8000/mcp
---
Option 2 — Run with Docker (Recommended)
Prerequisites: Docker, Docker Compose
git clone https://github.com/your-username/agenttrace.git
cd agenttrace
docker compose -f docker/docker-compose.yml up --buildThat's it. Docker will spin up PostgreSQL, wait for it to be healthy, run migrations automatically, and start the MCP server on port 8000.
Server runs at http://localhost:8000/mcp
---
Connect to VS Code
Once the server is running, add this to your VS Code MCP settings (Ctrl+Shift+P → Claude: Open MCP Settings):
{
"mcpServers": {
"agenttrace": {
"type": "http",
"url": "http://localhost:8000/mcp"
}
}
}Refresh MCP servers and AgentTrace will appear with all 7 tools available.
---
Connect to Any MCP-Compatible Agent
from fastmcp import Client
async with Client("http://localhost:8000/mcp") as client:
tools = await client.list_tools()
print(tools) # lists all 7 AgentTrace tools---
How It Works
AgentTrace uses three core concepts:
Session — the top-level container for a single run. Everything belongs to a session.
Span — a single unit of work inside a session. One agent calling another is one span. A human sending input is one span. Each span has a parent_span_id which is how AgentTrace builds the trace tree.
Event — something notable that happened inside a span. An error, a warning, or a log message.
Span Types
| Type | When to use | | ------------- | ------------------------------------------ | | human_input | A user sends a message to the system | | agent_call | One agent calls another agent | | tool_call | An agent invokes an MCP tool | | hitl | System pauses for human-in-the-loop review |
---
Tools Reference
create_session
Creates a new trace session. Call this at the start of every agent run.
Parameters:
| Parameter | Type | Required | Description | | ---------- | ------ | -------- | -------------------------------------------- | | name | string | ✅ | Human readable name for this run | | metadata | dict | ❌ | Optional key-value context e.g. env, version |
Returns: session_id, name, status, started_at, metadata
Example:
session = await client.call_tool("create_session", {
"name": "invoice-processing-run-42",
"metadata": {
"env": "production",
"triggered_by": "scheduler",
"version": "1.0.0"
}
})
session_id = session["session_id"]---
start_span
Opens a new span inside a session. Call this every time an agent starts a unit of work.
Parameters:
| Parameter | Type | Required | Description | | ---------------- | ------ | -------- | ------------------------------------------------------- | | session_id | string | ✅ | UUID of the parent session | | span_type | string | ✅ | One of human_input, agent_call, tool_call, hitl | | actor | string | ✅ | Who initiated — agent name or "human" | | target | string | ✅ | Who received — agent name or tool name | | input | dict | ❌ | Payload sent to the target | | parent_span_id | string | ❌ | UUID of parent span for nested calls |
Returns: span_id, session_id, parent_span_id, span_type, actor, target, status, started_at
Example:
# Top level span — human sends input
span1 = await client.call_tool("start_span", {
"session_id": session_id,
"span_type": "human_input",
"actor": "human",
"target": "orchestrator",
"input": {"message": "Process the invoice batch"}
})
# Child span — orchestrator calls an agent
span2 = await client.call_tool("start_span", {
"session_id": session_id,
"span_type": "agent_call",
"actor": "orchestrator",
"target": "invoice_agent",
"input": {"batch_id": "batch-99"},
"parent_span_id": span1["span_id"] # nested under span1
})---
end_span
Closes a span and records its output. Always call this when a unit of work finishes — success or failure.
Parameters:
| Parameter | Type | Required | Description | | --------- | ------ | -------- | -------------------------------- | | span_id | string | ✅ | UUID of the span to close | | status | string | ✅ | "success" or "failed" | | output | dict | ❌ | Payload returned from the target |
Returns: span_id, status, ended_at, output
Example:
# End successfully
await client.call_tool("end_span", {
"span_id": span2["span_id"],
"status": "success",
"output": {"invoices_processed": 42}
})
# End with failure
await client.call_tool("end_span", {
"span_id": span2["span_id"],
"status": "failed",
"output": {"error": "Batch not found"}
})---
log_event
Logs a notable event inside a span. Use this for errors, warnings, or anything worth capturing mid-span.
Parameters:
| Parameter | Type | Required | Description | | ------------ | ------ | -------- | -------------------------------------- | | span_id | string | ✅ | UUID of the span this event belongs to | | event_type | string | ✅ | One of log, error, warning | | message | string | ✅ | Human readable description | | metadata | dict | ❌ | Additional structured context |
Returns: event_id, span_id, event_type, message, created_at
Example:
# Log an error
await client.call_tool("log_event", {
"span_id": span2["span_id"],
"event_type": "error",
"message": "Invoice file not found on S3",
"metadata": {
"bucket": "invoices-prod",
"key": "batch-99/invoice.pdf",
"http_status": 404
}
})
# Log a warning
await client.call_tool("log_event", {
"span_id": span2["span_id"],
"event_type": "warning",
"message": "Retrying after rate limit",
"metadata": {"retry_attempt": 2, "wait_seconds": 5}
})---
complete_session / fail_session
Marks a session as completed or failed. Call one of these when the entire run finishes.
Parameters:
| Parameter | Type | Required | Description | | ------------ | ------ | -------- | ---------------------------- | | session_id | string | ✅ | UUID of the session to close |
Example:
# On success
await client.call_tool("complete_session", {"session_id": session_id})
# On failure
await client.call_tool("fail_session", {"session_id": session_id})---
query_session
Retrieves the full nested trace tree for a session. This is your primary debugging tool — one call gives you everything that happened in a run.
Parameters:
| Parameter | Type | Required | Description | | ------------ | ------ | -------- | ------------------------------- | | session_id | string | ✅ | UUID of the session to retrieve |
Returns: Full nested trace with all spans and events
Example:
trace = await client.call_tool("query_session", {
"session_id": session_id
})
# Response structure:
# {
# "session_id": "abc-123",
# "name": "invoice-processing-run-42",
# "status": "failed",
# "spans": [
# {
# "span_id": "...",
# "span_type": "human_input",
# "actor": "human",
# "target": "orchestrator",
# "status": "failed",
# "events": [],
# "children": [
# {
# "span_id": "...",
# "span_type": "agent_call",
# "status": "failed",
# "events": [
# {
# "event_type": "error",
# "message": "Invoice file not found on S3"
# }
# ],
# "children": []
# }
# ]
# }
# ]
# }---
query_cross_session
Queries spans across all sessions with filters. Use this to find recurring failures, slow agents, or patterns across multiple runs.
Parameters:
| Parameter | Type | Required | Description | | ---------------- | ------ | -------- | ------------------------------------------------------------------- | | actor | string | ❌ | Filter by actor name e.g. "invoice_agent" | | span_type | string | ❌ | Filter by type — human_input, agent_call, tool_call, hitl | | status | string | ❌ | Filter by status — running, success, failed | | event_type | string | ❌ | Filter spans containing this event type — log, error, warning | | started_after | string | ❌ | ISO datetime e.g. "2026-01-01T00:00:00" | | started_before | string | ❌ | ISO datetime e.g. "2026-12-31T23:59:59" |
Returns: List of matching spans with session context and events
Example:
# Find all failed spans from invoice_agent this week
results = await client.call_tool("query_cross_session", {
"actor": "invoice_agent",
"status": "failed",
"started_after": "2026-03-17T00:00:00"
})
# Find all spans that had an error event
results = await client.call_tool("query_cross_session", {
"event_type": "error"
})
# Find all tool_call spans that failed
results = await client.call_tool("query_cross_session", {
"span_type": "tool_call",
"status": "failed"
})---
replay_session
Creates a new session and returns all original inputs from a past session so agents can re-execute them fresh. Useful for reproducing bugs without manually re-running the entire system.
Parameters:
| Parameter | Type | Required | Description | | -------------- | ------ | -------- | -------------------------------------------------------------------------------- | | session_id | string | ✅ | UUID of the session to replay | | from_span_id | string | ❌ | Start replay from a specific span — useful to re-run only from the failure point |
Returns: new_session_id, replayed_from_session_id, spans_to_replay
Example:
# Full replay from the beginning
replay = await client.call_tool("replay_session", {
"session_id": session_id
})
# Partial replay — start from the failure point
replay = await client.call_tool("replay_session", {
"session_id": session_id,
"from_span_id": failed_span_id
})
# Feed inputs back to your agents using the new session
new_session_id = replay["new_session_id"]
for span in replay["spans_to_replay"]:
await your_agent.run(span["input"], session_id=new_session_id)---
Full Integration Example
Complete example showing how to wrap an existing multi-agent system with AgentTrace:
import asyncio
from fastmcp import Client
AGENTTRACE_URL = "http://localhost:8000/mcp"
async def run_pipeline(user_message: str):
async with Client(AGENTTRACE_URL) as tracer:
# 1. Start session
session = await tracer.call_tool("create_session", {
"name": "pipeline-run",
"metadata": {"triggered_by": "api"}
})
session_id = session["session_id"]
try:
# 2. Trace human input
root_span = await tracer.call_tool("start_span", {
"session_id": session_id,
"span_type": "human_input",
"actor": "human",
"target": "orchestrator",
"input": {"message": user_message}
})
# 3. Trace orchestrator → agent call
agent_span = await tracer.call_tool("start_span", {
"session_id": session_id,
"span_type": "agent_call",
"actor": "orchestrator",
"target": "research_agent",
"input": {"task": user_message},
"parent_span_id": root_span["span_id"]
})
try:
# Your actual agent logic here
result = await research_agent.run(user_message)
await tracer.call_tool("end_span", {
"span_id": agent_span["span_id"],
"status": "success",
"output": {"result": result}
})
except Exception as e:
await tracer.call_tool("log_event", {
"span_id": agent_span["span_id"],
"event_type": "error",
"message": str(e),
"metadata": {"exception_type": type(e).__name__}
})
await tracer.call_tool("end_span", {
"span_id": agent_span["span_id"],
"status": "failed"
})
raise
# 4. Close root span and complete session
await tracer.call_tool("end_span", {
"span_id": root_span["span_id"],
"status": "success"
})
await tracer.call_tool("complete_session", {
"session_id": session_id
})
except Exception:
await tracer.call_tool("fail_session", {"session_id": session_id})
raise
asyncio.run(run_pipeline("Summarize the quarterly report"))---
Query Examples
Debugging a Specific Run
Scenario: A user-reported issue with invoice processing. You remember the session ID from the error logs.
# Get the full trace tree to understand what went wrong
trace = await tracer.call_tool("query_session", {
"session_id": "69f4f019-b5ca-46a2-bf5e-59fb176c3227"
})
# Output shows the entire call chain with errors
# Navigate to the failed span and see exactly where it brokeFind All Errors This Week
Scenario: You want to see all errors across your system in the last 7 days.
results = await tracer.call_tool("query_cross_session", {
"event_type": "error",
"started_after": "2026-03-17T00:00:00",
"started_before": "2026-03-24T23:59:59"
})
# Returns all spans containing error events — useful for weekly reviewsSpot Failing Agents
Scenario: The invoice_agent has been failing. Find all its failures.
results = await tracer.call_tool("query_cross_session", {
"actor": "invoice_agent",
"status": "failed"
})
# See every failed run from this agent — identify patternsFind Slow Tool Calls
Scenario: Tool calls are taking too long. Find all tool calls that might be problematic.
results = await tracer.call_tool("query_cross_session", {
"span_type": "tool_call",
"status": "success"
})
# Returns all successful tool calls — you can inspect duration and optimizeReproduce a Bug
Scenario: A user reports a bug. Replay the session to debug locally.
# Find the failed session
failed_session = await tracer.call_tool("query_cross_session", {
"actor": "research_agent",
"status": "failed",
"started_after": "2026-03-24T12:00:00"
})
session_id = failed_session[0]["session_id"]
# Replay it in a fresh session with new logic
replay = await tracer.call_tool("replay_session", {
"session_id": session_id
})
new_session = replay["new_session_id"]
# Re-run your pipeline with the new session ID — same inputs, fresh codeFind All Pending HITL Reviews
Scenario: You want to see all human-in-the-loop approvals that are still pending.
results = await tracer.call_tool("query_cross_session", {
"span_type": "hitl",
"status": "running"
})
# See all pending approvals across all sessionsCross-Session Pattern Analysis
Scenario: Analyze all agent-to-agent calls to identify bottlenecks.
results = await tracer.call_tool("query_cross_session", {
"span_type": "agent_call"
})
# Returns all inter-agent calls — analyze metrics like success rate, frequency
for span in results:
print(f"{span['actor']} → {span['target']}: {span['status']}")---
Project Structure
mcp-tracer/
├── src/
│ └── mcp_tracer/
│ ├── server.py # FastMCP entry point
│ ├── tools/ # MCP tool handlers
│ ├── db/ # Models + repositories
│ ├── schemas/ # Pydantic I/O schemas
│ ├── services/ # Business logic
│ └── core/ # Config, logging, exceptions
├── migrations/ # Alembic migrations
├── tests/
└── docker/---
Roadmap
v1 (Current)
- [x] Session lifecycle management
- [x] Span tracking with parent-child nesting
- [x] Event logging
- [x] Full trace tree reconstruction
- [x] Cross-session querying with filters
- [x] Soft session replay
v2 (Planned)
- [ ] Hard replay — snapshot and re-inject tool outputs for identical reproduction
- [ ] Visual DAG dashboard
- [ ] Log sampling and levels for production
- [ ] Session branching for retries
- [ ] Alerting hooks
- [ ] Agent confidence and anomaly scoring
---
Contributing
First off, thank you for considering contributing to #AgnetTrace! It's people like you that make the open-source community such an amazing place to learn, inspire, and create.
We welcome contributions of all kinds, whether it is fixing bugs, improving documentation, suggesting new features, or writing code.
---
⚖️ License and Contributions
This project is open-source and released under the MIT License. By contributing to this repository, you agree that your contributions will be licensed under its MIT License.
---
🚀 How Can I Contribute?
Reporting Bugs
If you find a bug, please create an issue to report it. To help us resolve it quickly, include:
- A clear and descriptive title.
- Steps to reproduce the exact behavior.
- Expected behavior versus what actually happened.
- Your operating system, browser, and relevant version numbers.
Suggesting Enhancements
Feature requests are always welcome! When proposing a new feature, please open an issue and include:
- The problem this feature solves.
- A detailed description of the proposed solution.
- Any alternative solutions you have considered.
- Mockups or code snippets if applicable.
---
🛠️ Pull Request Process
When you are ready to submit your code, follow this process:
- Create a new branch for your feature or bugfix:
git checkout -b feature/your-feature-name - Make your changes and test them thoroughly.
- Commit your changes with clear, descriptive commit messages.
- Push your branch to your forked repository:
git push origin feature/your-feature-name - Open a Pull Request (PR) against the
feature/External-PRbranch of the original repository. - Provide a detailed description in your PR explaining what changes you made and why.
A maintainer will review your PR, provide feedback, and merge it once it is approved.
---
🤝 Code of Conduct
Please note that this project is released with a Contributor Code of Conduct. By participating in this project, you agree to abide by its terms. We expect all contributors to maintain a respectful and welcoming environment for everyone.
