multiagent-trace-middleware 0.1.0

Observability middleware for multi-agent LangGraph/LangChain workflows. Track agent trees, emit events, and trace execution across nested agents.

multiagent-trace-middleware

Real-time tracing middleware for Python agents. Build terminal UIs or desktop apps that display your LangGraph/LangChain agent execution with hierarchical visualization, color-coded agents, and live status updates.

What is this for?

When building apps that integrate Python AI agents (LangGraph, LangChain, etc.), you often need to display what your agent is doing in real-time. This middleware hooks into your agent and emits structured events that your UI can consume to show:

• Which agent is currently running (with customizable colors for each agent)
• Hierarchical nesting levels (main agent → subagent → sub-subagent)
• Tool calls and their results (with arguments and completion status)
• Agent thinking/reasoning in real-time
• Todo list updates as agents work through tasks

Perfect for building:

• Terminal-based agent UIs (like the screenshots above)
• Desktop applications with agent monitoring
• Web dashboards showing agent execution
• Debugging tools for multi-agent systems

Features

• Agent Hierarchy Tracking: Automatically tracks parent-child relationships between agents
• Event Emission: Emits structured events for thinking, tool calls, tool results, and completions
• Dual Output: Supports both EventBus (programmatic) and JSON stdout (terminal apps)
• Color Support: Customizable foreground/background colors per agent for UI display
• Level Tracking: Automatic nesting level calculation (0=root, 1=subagent, etc.)

Installation

pip install multiagent-trace-middleware

Or with uv:

uv add multiagent-trace-middleware

Quick Start

from multiagent_trace_middleware import AgentObserverMiddleware, get_event_bus

# Add middleware to your main agent
middleware = [
    AgentObserverMiddleware(
        "main",
        fg_color="#FFFFFF",
        bg_color="#1E88E5"
    ),
]

# Add middleware to subagents
subagent_middleware = [
    AgentObserverMiddleware(
        "researcher",
        fg_color="#000000",
        bg_color="#4CAF50"
    ),
]

# Subscribe to events
def on_event(event_type, agent_name, fg_color, bg_color, level, data):
    print(f"[L{level}] {agent_name}: {event_type}")

get_event_bus().subscribe(on_event)

Event Types

Event Type	Description
agent_change	Emitted when switching between agents
thinking	AI reasoning/content output
tool_call	Tool invocation with arguments
tool_result	Tool execution results
completion	Agent completion messages
todos	Todo list updates
message	Raw message objects (EventBus only)

JSON Output Format

When emit_json=True (default), events are written to stdout as JSON:

{
  "type": "tool_call",
  "timestamp": "2024-01-15T10:30:00.000000",
  "agent_name": "researcher",
  "agent_level": 1,
  "agent_fg_color": "#000000",
  "agent_bg_color": "#4CAF50",
  "tool_name": "search",
  "tool_call_id": "call_abc123",
  "args": {"query": "example"}
}

API Reference

AgentObserverMiddleware

AgentObserverMiddleware(
    name: str,                    # Agent name for display
    fg_color: str = "#000000",    # Foreground color (hex)
    bg_color: str = "#ffffff",    # Background color (hex)
    emit_json: bool = True,       # Emit JSON to stdout
    emit_events: bool = True,     # Emit to EventBus
)

EventBus

from multiagent_trace_middleware import get_event_bus

bus = get_event_bus()
bus.subscribe(callback)      # Subscribe to events
bus.unsubscribe(callback)    # Unsubscribe
bus.clear()                  # Clear all listeners

Agent Tree

from multiagent_trace_middleware import get_agent_tree, reset_agent_tree

tree = get_agent_tree()
node = tree.current          # Get current agent node
level = tree.current_level   # Get current nesting level
reset_agent_tree()           # Reset tree (useful for testing)

License

MIT