fix: ManagedAgentGraph takes config as first required parameter

Aligns ManagedAgentGraph constructor with the ManagedModel and ManagedAgent
pattern: graph (AgentGraphDefinition) is now the first required parameter,
followed by the runner. Removes the Optional guard and tightens the type
annotation accordingly. Updates client.py call site and all test fixtures.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

Author: jsonbailey

packages/sdk/server-ai/src/ldai/client.py

@@ -799,7 +799,7 @@ async def create_agent_graph(
         if not runner:
             return None
 
-        return ManagedAgentGraph(runner, graph=graph)
+        return ManagedAgentGraph(graph, runner)
 
     def agents(
         self,

packages/sdk/server-ai/src/ldai/managed_agent_graph.py

@@ -1,7 +1,8 @@
 """ManagedAgentGraph — LaunchDarkly managed wrapper for agent graph execution."""
 
-from typing import Any, Optional
+from typing import Any
 
+from ldai.agent_graph import AgentGraphDefinition
 from ldai.providers import AgentGraphRunner
 from ldai.providers.types import (
     AgentGraphRunnerResult,
@@ -23,18 +24,18 @@ class ManagedAgentGraph:
 
     def __init__(
         self,
+        graph: AgentGraphDefinition,
         runner: AgentGraphRunner,
-        graph: Optional[Any] = None,
     ):
         """
         Initialize ManagedAgentGraph.
 
-        :param runner: The AgentGraphRunner to delegate execution to
-        :param graph: Optional AgentGraphDefinition used to drive graph-level and
+        :param graph: The AgentGraphDefinition used to drive graph-level and
             per-node tracking from the runner result metrics.
+        :param runner: The AgentGraphRunner to delegate execution to
         """
-        self._runner = runner
         self._graph = graph
+        self._runner = runner
 
     async def run(self, input: Any) -> ManagedGraphResult:
         """
@@ -56,10 +57,9 @@ async def run(self, input: Any) -> ManagedGraphResult:
 
         summary = self._build_summary_from_runner_result(result)
 
-        if self._graph is not None:
-            graph_tracker = self._graph.create_tracker()
-            self._flush_graph_tracking(result, graph_tracker)
-            self._flush_node_tracking(result)
+        graph_tracker = self._graph.create_tracker()
+        self._flush_graph_tracking(result, graph_tracker)
+        self._flush_node_tracking(result)
 
         return ManagedGraphResult(
             content=result.content,
@@ -106,9 +106,6 @@ def _flush_node_tracking(self, result: AgentGraphRunnerResult) -> None:
         config tracker via the graph definition and fires token, duration,
         tool call, and success/error events.
         """
-        if self._graph is None:
-            return
-
         for node_key, node_ldai_metrics in result.metrics.node_metrics.items():
             node = self._graph.get_node(node_key)
             if node is None:

packages/sdk/server-ai/tests/test_managed_agent_graph.py

@@ -62,7 +62,10 @@ async def run(self, input) -> AgentGraphRunnerResult:
 async def test_managed_agent_graph_run_delegates_to_runner():
     """Runner result content is surfaced correctly."""
     runner = StubAgentGraphRunner("hello world")
-    managed = ManagedAgentGraph(runner)
+    mock_graph = MagicMock()
+    mock_graph.create_tracker = MagicMock(return_value=MagicMock())
+    mock_graph.get_node = MagicMock(return_value=None)
+    managed = ManagedAgentGraph(mock_graph, runner)
     result = await managed.run("test input")
     assert isinstance(result, ManagedGraphResult)
     assert result.content == "hello world"
@@ -71,7 +74,8 @@ async def test_managed_agent_graph_run_delegates_to_runner():
 
 def test_managed_agent_graph_get_runner():
     runner = StubAgentGraphRunner()
-    managed = ManagedAgentGraph(runner)
+    mock_graph = MagicMock()
+    managed = ManagedAgentGraph(mock_graph, runner)
     assert managed.get_agent_graph_runner() is runner
 
 
@@ -84,7 +88,7 @@ async def test_managed_agent_graph_run_surfaces_graph_metrics():
     mock_graph.create_tracker = MagicMock(return_value=mock_tracker)
     mock_graph.get_node = MagicMock(return_value=None)  # no nodes for this test
 
-    managed = ManagedAgentGraph(runner, graph=mock_graph)
+    managed = ManagedAgentGraph(mock_graph, runner)
     result = await managed.run("test input")
 
     assert isinstance(result, ManagedGraphResult)
@@ -105,7 +109,7 @@ async def test_managed_agent_graph_drives_graph_level_tracking():
     mock_graph.create_tracker = MagicMock(return_value=mock_tracker)
     mock_graph.get_node = MagicMock(return_value=None)
 
-    managed = ManagedAgentGraph(runner, graph=mock_graph)
+    managed = ManagedAgentGraph(mock_graph, runner)
     await managed.run("test input")
 
     mock_tracker.track_path.assert_called_once_with(["root", "specialist"])
@@ -135,7 +139,7 @@ def get_node(key):
 
     mock_graph.get_node = get_node
 
-    managed = ManagedAgentGraph(runner, graph=mock_graph)
+    managed = ManagedAgentGraph(mock_graph, runner)
     await managed.run("test input")
 
     # root node tracking
@@ -150,10 +154,14 @@ def get_node(key):
 
 
 @pytest.mark.asyncio
-async def test_managed_agent_graph_no_graph_skips_tracking():
-    """Without a graph reference, no tracking is called but run succeeds."""
+async def test_managed_agent_graph_run_succeeds_with_graph():
+    """Run succeeds and returns correct content when graph is provided."""
     runner = StubRunnerWithMetrics()
-    managed = ManagedAgentGraph(runner, graph=None)
+    mock_graph = MagicMock()
+    mock_tracker = MagicMock()
+    mock_graph.create_tracker = MagicMock(return_value=mock_tracker)
+    mock_graph.get_node = MagicMock(return_value=None)
+    managed = ManagedAgentGraph(mock_graph, runner)
     result = await managed.run("test input")
     assert result.content == "new shape output"
     assert result.metrics.success is True
@@ -176,7 +184,7 @@ async def run(self, input) -> AgentGraphRunnerResult:
     mock_graph.create_tracker = MagicMock(return_value=mock_tracker)
     mock_graph.get_node = MagicMock(return_value=None)
 
-    managed = ManagedAgentGraph(FailingRunner(), graph=mock_graph)
+    managed = ManagedAgentGraph(mock_graph, FailingRunner())
     result = await managed.run("test input")
 
     assert result.metrics.success is False