Monitor AutoGen Workflows with Nexus

Step 1 — Install the SDK

pip install keylightdigital-nexus autogen-agentchat

Step 2 — Create an API key

Go to /dashboard/keys and create a new API key. Store it as NEXUS_API_KEY.

Step 3 — Wrap your AutoGen workflow

from autogen import AssistantAgent, UserProxyAgent
from nexus_client import NexusClient
import os

nexus = NexusClient(
    api_key=os.environ["NEXUS_API_KEY"],
    agent_id="autogen-research-workflow",
)

config_list = [
    {
        "model": "gpt-4o",
        "api_key": os.environ["OPENAI_API_KEY"],
    }
]

# Create AutoGen agents
assistant = AssistantAgent(
    name="assistant",
    llm_config={"config_list": config_list},
)

user_proxy = UserProxyAgent(
    name="user_proxy",
    human_input_mode="NEVER",
    max_consecutive_auto_reply=5,
    code_execution_config={"work_dir": "coding", "use_docker": False},
)

def run_workflow(task: str) -> None:
    # Start Nexus trace for the full conversation
    trace = nexus.start_trace(
        name=f"AutoGen: {task[:60]}",
        metadata={"task": task, "agents": ["assistant", "user_proxy"]},
    )

    try:
        # Track the initiation
        trace.add_span(
            name="workflow-start",
            input={"task": task, "agent_count": 2},
        )

        # Run the AutoGen conversation
        user_proxy.initiate_chat(
            assistant,
            message=task,
        )

        # Log conversation summary
        history = user_proxy.chat_messages.get(assistant, [])
        trace.add_span(
            name="workflow-complete",
            output={
                "message_count": len(history),
                "last_role": history[-1]["role"] if history else None,
            },
        )

        trace.end(status="success")

    except Exception as e:
        trace.add_span(
            name="workflow-error",
            error=str(e),
        )
        trace.end(status="error")
        raise


if __name__ == "__main__":
    run_workflow("Write and test a Python function that checks if a number is prime")

Advanced — per-turn spans with a reply hook

For finer-grained visibility, hook into AutoGen's message passing to record each agent turn:

from autogen import AssistantAgent, UserProxyAgent, Agent
from nexus_client import NexusClient, Trace
import os

nexus = NexusClient(
    api_key=os.environ["NEXUS_API_KEY"],
    agent_id="autogen-workflow",
)

class TracedUserProxy(UserProxyAgent):
    """UserProxyAgent that records each turn to Nexus."""

    def __init__(self, *args, nexus_trace: Trace, **kwargs):
        super().__init__(*args, **kwargs)
        self._nexus_trace = nexus_trace
        self._turn = 0

    def receive(self, message, sender: Agent, request_reply=None, silent=False):
        self._turn += 1
        self._nexus_trace.add_span(
            name=f"turn-{self._turn}-{sender.name}",
            input={"role": sender.name, "turn": self._turn},
            output={"content": str(message)[:500]},
        )
        return super().receive(message, sender, request_reply, silent)


def run_traced_workflow(task: str) -> None:
    trace = nexus.start_trace(name=f"AutoGen: {task[:60]}")

    assistant = AssistantAgent(
        name="assistant",
        llm_config={"config_list": [{"model": "gpt-4o", "api_key": os.environ["OPENAI_API_KEY"]}]},
    )
    user_proxy = TracedUserProxy(
        name="user_proxy",
        nexus_trace=trace,
        human_input_mode="NEVER",
        max_consecutive_auto_reply=5,
    )

    try:
        user_proxy.initiate_chat(assistant, message=task)
        trace.end(status="success")
    except Exception:
        trace.end(status="error")
        raise

Next steps

• API Reference — full REST API documentation
• Interactive demo — see sample traces without signing up
• CrewAI guide — another multi-agent framework
• Blog: Building Autonomous Agents with Observability
• Nexus pricing — free plan or $9/mo Pro
• GitHub — open-source SDK