Beyond n8n for workflow automation: agent graphs as your universal agent harness

# LangGraph - topology hardcoded in graph setup
workflow = StateGraph(AgentState)
workflow.add_node("supervisor", supervisor_node)
workflow.add_node("security", security_node)
workflow.add_node("support", support_node)
workflow.set_entry_point("supervisor")
# Routing logic buried in node functions or conditional edges

# OpenAI Agents SDK - handoffs defined per agent
security_agent = Agent(name="Security", instructions="...")
support_agent = Agent(name="Support", instructions="...")
supervisor = Agent(
    name="Supervisor",
    handoffs=[security_agent, support_agent]  # Topology locked in code
)

{
  "action": "sanitize",
  "reason": "PII detected",
  "route": "security"
}

{
  "action": "direct",
  "reason": "Clean input",
  "route": "support"
}

{
  "action": "proceed",
  "reason": "Input sanitized",
  "route": "continue"
}

uv add launchdarkly-server-sdk launchdarkly-server-sdk-ai

# config_manager.py - Initialize LaunchDarkly clients
def _initialize_launchdarkly_client(self):
    """Initialize LaunchDarkly client and AI client"""
    config = ldclient.Config(self.sdk_key)
    ldclient.set_config(config)
    self.ld_client = ldclient.get()

    # Block until client is initialized (max 10 seconds)
    self.ld_client.start_wait(10)

    if not self.ld_client.is_initialized():
        raise RuntimeError("LaunchDarkly client initialization failed")

    self.ai_client = LDAIClient(self.ld_client)

# config_manager.py - Build context for targeting
def build_context(self, user_id: str, user_context: dict = None) -> Context:
    """Build a LaunchDarkly context with consistent attributes."""
    context_builder = Context.builder(user_id).kind('user')

    if user_context:
        for key, value in user_context.items():
            context_builder.set(key, value)

    return context_builder.build()

# agents/generic_agent.py
def create_generic_agent(agent_config, config_manager, valid_routes: List[str] = None):
    """Create a generic agent from a config."""

    class GenericAgent:
        def __init__(self):
            self.valid_routes = valid_routes or []

        async def ainvoke(self, state: dict) -> dict:
            """Execute the agent using config."""
            if not agent_config.enabled:
                return {"response": "", "_skipped": True}

            # Create model from config
            model = create_model_for_config(
                provider=agent_config.provider.name,
                model=agent_config.model.name,
                config_manager=config_manager
            )

            # Load tools from config
            tools = create_dynamic_tools_from_launchdarkly(agent_config)

            # Get instructions from config
            instructions = agent_config.instructions or "Process the input."

            # Inject route options into instructions
            if self.valid_routes:
                route_instruction = f"\n\nSelect one of these routes: {self.valid_routes}. Return: {{\"route\": \"<selected_route>\"}}"
                instructions = instructions + route_instruction

            # Execute and extract routing decision
            result = await self._execute(model, instructions, tools, state)
            result["routing_decision"] = self._extract_route(result.get("response", ""))

            tracker = agent_config.create_tracker()
            tracker.track_success()
            return result

    return GenericAgent()

# api/services/agent_service.py
class AgentService:
    """Multi-Agent Orchestration using LaunchDarkly Agent Graph."""

    def __init__(self):
        self.config_manager = ConfigManager()
        self.config_manager.flush()

    async def process_message(
        self,
        user_id: str,
        message: str,
        user_context: dict = None
    ) -> ChatResponse:
        """Process message using LaunchDarkly Agent Graph."""
        result = await self._execute_graph(
            graph_key=os.getenv("AGENT_GRAPH_KEY", "chatbot-flow"),
            user_id=user_id.strip() or "anonymous",
            user_input=message,
            user_context=user_context or {}
        )

        return ChatResponse(
            response=result.get("final_response", ""),
            tool_calls=result.get("tool_calls", []),
            # ... other fields
        )

# api/services/agent_service.py
async def _execute_graph(
    self,
    graph_key: str,
    user_id: str,
    user_input: str,
    user_context: dict = None
) -> Dict[str, Any]:
    """Execute agents using SDK's traverse() with skip logic."""
    ld_context = self.config_manager.build_context(user_id, user_context)
    graph = self.config_manager.ai_client.agent_graph(graph_key, ld_context)

    if not graph.is_enabled():
        raise ValueError(f"Agent Graph '{graph_key}' is not enabled")

    ctx = {
        "user_input": user_input,
        "messages": [HumanMessage(content=user_input)],
        "processed_input": user_input,
        "final_response": "",
        "tool_calls": [],
        # Skip logic: track which nodes should execute
        "_routed_to": {graph.root().get_key()},
        "_path": [],
        "_prev_key": None,
    }

    tracker = graph.create_tracker()

    # Define the node callback (see next section)
    def execute_node(node, exec_ctx):
        # ... node execution logic
        pass

    # Use SDK's traverse() - it handles traversal order
    graph.traverse(execute_node, ctx)

    # Track graph completion
    if tracker:
        tracker.track_path(ctx.get("_path", []))
        tracker.track_invocation_success()

    return ctx

# api/services/agent_service.py - inside _execute_graph
def execute_node(node, exec_ctx):
    """Execute a single node if it was routed to."""
    key = node.get_key()

    # Skip logic: only execute if parent routed to this node
    if key not in exec_ctx.get("_routed_to", set()):
        return {"_skipped": True}

    exec_ctx["_path"].append(key)

    node_tracker = node.get_config().create_tracker()
    node_tracker.track_success()
    if tracker and exec_ctx.get("_prev_key"):
        tracker.track_handoff_success(exec_ctx["_prev_key"], key)

    # Get edges and valid routes for this node
    edges = node.get_edges()
    valid_routes = [e.handoff.get("route") for e in edges if e.handoff and e.handoff.get("route")]

    # Execute agent with config from this node
    agent = create_generic_agent(node.get_config(), self.config_manager, valid_routes=valid_routes)
    result = _run_async(agent.ainvoke(exec_ctx))

    # Track tool calls
    if result.get("tool_calls"):
        for tool in result["tool_calls"]:
            node_tracker.track_tool_call(tool)

    # Route to next node: add to _routed_to set
    if edges:
        next_key = self._select_next_node(node, edges, result, tracker)
        if next_key:
            exec_ctx["_routed_to"].add(next_key)

    exec_ctx["_prev_key"] = key
    return result

# api/services/agent_service.py
def _select_next_node(self, node, edges, result: dict, tracker=None):
    """Select next node key based on routing decision."""
    routing = result.get("routing_decision", "").lower().strip() if result.get("routing_decision") else None

    # Build route map: route -> target_config
    route_map = {}
    for edge in edges:
        route = (edge.handoff.get("route", "") if edge.handoff else "").lower().strip()
        if route:
            route_map[route] = edge.target_config

    # Exact match
    if routing and routing in route_map:
        return route_map[routing]
    elif routing:
        if tracker:
            target = edges[0].target_config if edges else None
            tracker.track_handoff_failure(node.get_key(), target)

    # Default: first edge
    if edges:
        return edges[0].target_config

    return None

service = AgentService()
response = await service.process_message(
    user_id="user-123",
    message="What's the status of my order?",
    user_context={"plan": "premium"}
)
print(response.response)

uv run python tools/traffic_generator.py --queries 20 --delay 2

# Conceptual example showing how Agent Graph SDK methods work with OpenAI Agents
from agents import Agent, Runner

def handle_traversal(node, state):
    config = node.get_config()
    tracker = config.create_tracker()
    edges = node.get_edges()

    # Child agents are already in state (reverse traversal builds bottom-up)
    handoffs = [state[edge.target_config] for edge in edges]

    def on_handoff(ctx):
        # Track handoff events
        return ctx

    return Agent(
        name=config.key,
        instructions=config.instructions,
        handoffs=handoffs,
        on_handoff=on_handoff,
    )

if agent_graph.is_enabled():
    root = agent_graph.reverse_traverse(handle_traversal, {})

result = await Runner.run(root, "Tell me about your engineering team")