Skip to content

LangChain/LangGraph middleware for tracing multi-agent execution. Tracks agent hierarchy, emits structured events, and provides real-time observability into nested agent workflows.

License

Notifications You must be signed in to change notification settings

atom2ueki/multiagent-trace-middleware

Repository files navigation

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.

Main Agent View Multi-Agent Hierarchy

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

About

LangChain/LangGraph middleware for tracing multi-agent execution. Tracks agent hierarchy, emits structured events, and provides real-time observability into nested agent workflows.

Resources

License

Stars

Watchers

Forks

Packages

No packages published

Languages