feat(sdk): add LangGraph ToolNode integration

examples/langchain/langchain_middleware_smoke.py

@@ -0,0 +1,104 @@
+"""Minimal LangChain middleware smoke test for Agent Control.
+
+This example proves the LangChain agent-factory middleware path works with
+``AgentControlMiddleware`` and protects tool calls without using ``@control()``.
+It passes a plain ``@tool``-decorated Python function directly to
+``langchain.create_agent(...)``.
+
+Run:
+    cd examples/langchain
+    uv run setup_langchain_middleware_controls.py
+    uv run langchain_middleware_smoke.py
+
+Prerequisite:
+    Start the Agent Control server first (`cd server && make run`).
+"""
+
+from __future__ import annotations
+
+import os
+
+import agent_control
+from agent_control.integrations.langchain import AgentControlMiddleware
+from langchain.agents import create_agent
+from langchain_core.language_models import FakeMessagesListChatModel
+from langchain_core.messages import AIMessage, HumanMessage, ToolMessage
+from langchain_core.tools import tool
+
+AGENT_NAME = "langchain-middleware-smoke"
+AGENT_DESCRIPTION = "Minimal LangChain middleware smoke test using Agent Control"
+
+
+@tool("get_weather")
+def get_weather(city: str) -> dict[str, str]:
+    """Return a deterministic weather response for a city."""
+    return {
+        "city": city,
+        "forecast": {
+            "seattle": "Rainy and 53F",
+            "tehran": "Sunny and 75F",
+            "tokyo": "Clear and 61F",
+        }.get(city.lower(), "Partly cloudy and 68F"),
+    }
+
+
+def _build_agent_for_city(city: str):
+    tool_call_id = f"call-weather-{city.lower().replace(' ', '-')}"
+    model = FakeMessagesListChatModel(
+        responses=[
+            AIMessage(
+                content="",
+                tool_calls=[
+                    {
+                        "name": "get_weather",
+                        "args": {"city": city},
+                        "id": tool_call_id,
+                        "type": "tool_call",
+                    }
+                ],
+            ),
+            AIMessage(content=f"Finished processing weather lookup for {city}."),
+        ]
+    )
+    return create_agent(
+        model=model,
+        tools=[get_weather],
+        middleware=[AgentControlMiddleware()],
+        system_prompt="Always use the available weather tool before replying.",
+        name="agent-control-langchain-middleware-smoke",
+    )
+
+
+def _run_scenario(prompt: str, city: str) -> None:
+    app = _build_agent_for_city(city)
+    result = app.invoke({"messages": [HumanMessage(content=prompt)]})
+
+    print("=" * 80)
+    print(f"User: {prompt}")
+    print(f"Final response: {result['messages'][-1].content}")
+
+    tool_message = next(
+        message for message in reversed(result["messages"]) if isinstance(message, ToolMessage)
+    )
+    print(
+        f"Raw tool message -> name={tool_message.name!r}, "
+        f"status={getattr(tool_message, 'status', None)!r}, "
+        f"content={tool_message.content!r}"
+    )
+
+
+def main() -> None:
+    """Run the example with one allowed and one blocked scenario."""
+    agent_control.init(
+        agent_name=AGENT_NAME,
+        agent_description=AGENT_DESCRIPTION,
+        server_url=os.getenv("AGENT_CONTROL_URL"),
+    )
+
+    print("Running LangChain middleware smoke test...")
+    _run_scenario("What is the weather in Seattle?", "Seattle")
+    _run_scenario("What is the weather in Tehran?", "Tehran")
+
+
+if __name__ == "__main__":
+    main()

examples/langchain/langgraph_toolnode_integration_smoke.py

@@ -0,0 +1,138 @@
+"""Minimal LangGraph smoke test for the Agent Control ToolNode integration.
+
+This example proves the LangGraph ToolNode wrapper path works without using
+``@control()`` on the underlying tool implementation. It passes a plain
+``@tool``-decorated Python function directly to the Agent Control LangGraph
+integration.
+
+Run:
+    cd examples/langchain
+    uv run setup_langgraph_toolnode_controls.py
+    uv run langgraph_toolnode_integration_smoke.py
+
+Prerequisite:
+    Start the Agent Control server first (`cd server && make run`).
+"""
+
+from __future__ import annotations
+
+import asyncio
+import os
+import re
+from typing import Annotated, TypedDict
+
+import agent_control
+from agent_control.integrations.langgraph import create_controlled_tool_node
+from langchain_core.messages import AIMessage, BaseMessage, HumanMessage, ToolMessage
+from langchain_core.tools import tool
+from langgraph.graph import END, START, StateGraph
+from langgraph.graph.message import add_messages
+
+AGENT_NAME = "langgraph-toolnode-smoke"
+AGENT_DESCRIPTION = "Minimal LangGraph ToolNode smoke test using Agent Control integration"
+
+
+class AgentState(TypedDict):
+    """LangGraph state object."""
+
+    messages: Annotated[list[BaseMessage], add_messages]
+
+
+@tool("get_weather")
+async def get_weather(city: str) -> dict[str, str]:
+    """Return a deterministic weather response for a city."""
+    return {
+        "city": city,
+        "forecast": {
+            "seattle": "Rainy and 53F",
+            "tehran": "Sunny and 75F",
+            "tokyo": "Clear and 61F",
+        }.get(city.lower(), "Partly cloudy and 68F"),
+    }
+
+
+def _extract_city(user_text: str) -> str:
+    """Extract a city token from a simple prompt."""
+    match = re.search(r"(?:for|in)\s+([A-Za-z][A-Za-z\\s-]*)", user_text)
+    if match:
+        return match.group(1).strip().rstrip("?.!")
+    return user_text.strip().split()[-1].rstrip("?.!")
+
+
+def _build_graph():
+    """Build a deterministic graph that always routes to the weather tool."""
+    tool_node = create_controlled_tool_node([get_weather])
+
+    def planner(state: AgentState) -> dict[str, list[AIMessage]]:
+        user_text = str(state["messages"][-1].content)
+        city = _extract_city(user_text)
+        tool_call = {
+            "name": "get_weather",
+            "args": {"city": city},
+            "id": f"call-weather-{city.lower().replace(' ', '-')}",
+            "type": "tool_call",
+        }
+        return {"messages": [AIMessage(content="", tool_calls=[tool_call])]}  # type: ignore[arg-type]
+
+    def finalize(state: AgentState) -> dict[str, list[AIMessage]]:
+        tool_message = next(
+            message for message in reversed(state["messages"]) if isinstance(message, ToolMessage)
+        )
+        status = getattr(tool_message, "status", None) or "success"
+        return {
+            "messages": [
+                AIMessage(
+                    content=(
+                        f"Tool `{tool_message.name}` finished with status `{status}`: "
+                        f"{tool_message.content}"
+                    )
+                )
+            ]
+        }
+
+    graph = StateGraph(AgentState)
+    graph.add_node("planner", planner)
+    graph.add_node("tools", tool_node)
+    graph.add_node("finalize", finalize)
+    graph.add_edge(START, "planner")
+    graph.add_edge("planner", "tools")
+    graph.add_edge("tools", "finalize")
+    graph.add_edge("finalize", END)
+    return graph.compile()
+
+
+async def main() -> None:
+    """Run the example with one allowed and one blocked scenario."""
+    agent_control.init(
+        agent_name=AGENT_NAME,
+        agent_description=AGENT_DESCRIPTION,
+        server_url=os.getenv("AGENT_CONTROL_URL"),
+    )
+
+    app = _build_graph()
+    scenarios = [
+        "What is the weather in Seattle?",
+        "What is the weather in Tehran?",
+    ]
+
+    print("Running LangGraph ToolNode integration smoke test...")
+    for prompt in scenarios:
+        print("=" * 80)
+        print(f"User: {prompt}")
+        result = await app.ainvoke({"messages": [HumanMessage(content=prompt)]})
+
+        final_message = result["messages"][-1]
+        print(final_message.content)
+
+        tool_message = next(
+            message for message in reversed(result["messages"]) if isinstance(message, ToolMessage)
+        )
+        print(
+            f"Raw tool message -> name={tool_message.name!r}, "
+            f"status={getattr(tool_message, 'status', None)!r}, "
+            f"content={tool_message.content!r}"
+        )
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

examples/langchain/setup_langchain_middleware_controls.py

@@ -0,0 +1,108 @@
+"""Create controls for the LangChain middleware smoke test.
+
+This script prepares a single direct agent control for the
+``langchain_middleware_smoke.py`` example.
+
+Run:
+    cd examples/langchain
+    uv run setup_langchain_middleware_controls.py
+"""
+
+from __future__ import annotations
+
+import asyncio
+import os
+from typing import Any
+
+import httpx
+from agent_control import Agent, AgentControlClient, agents, controls
+
+AGENT_NAME = "langchain-middleware-smoke"
+SERVER_URL = os.getenv("AGENT_CONTROL_URL", "http://localhost:8000")
+
+CONTROL_SPECS: list[tuple[str, dict[str, Any]]] = [
+    (
+        "langchain-middleware-block-city",
+        {
+            "description": "Block restricted cities before the get_weather tool runs.",
+            "enabled": True,
+            "execution": "server",
+            "scope": {
+                "step_types": ["tool"],
+                "step_names": ["get_weather"],
+                "stages": ["pre"],
+            },
+            "selector": {"path": "input.city"},
+            "evaluator": {
+                "name": "list",
+                "config": {
+                    "values": ["Tehran", "Pyongyang"],
+                    "logic": "any",
+                    "match_on": "match",
+                    "match_mode": "exact",
+                    "case_sensitive": False,
+                },
+            },
+            "action": {
+                "decision": "deny",
+                "message": "That city is blocked by policy.",
+            },
+        },
+    ),
+]
+
+
+async def _ensure_control(
+    client: AgentControlClient,
+    name: str,
+    data: dict[str, Any],
+) -> int:
+    """Create a control or update the existing definition."""
+    try:
+        result = await controls.create_control(client, name=name, data=data)
+        return int(result["control_id"])
+    except httpx.HTTPStatusError as exc:
+        if exc.response.status_code != 409:
+            raise
+
+    control_list = await controls.list_controls(client, name=name, limit=1)
+    existing = control_list.get("controls", [])
+    if not existing:
+        raise RuntimeError(f"Control '{name}' already exists but could not be listed.")
+
+    control_id = int(existing[0]["id"])
+    await controls.set_control_data(client, control_id, data)
+    return control_id
+
+
+async def main() -> None:
+    """Register the example agent and ensure its controls exist."""
+    async with AgentControlClient(base_url=SERVER_URL) as client:
+        await client.health_check()
+
+        agent = Agent(
+            agent_name=AGENT_NAME,
+            agent_description="LangChain middleware smoke test using Agent Control",
+        )
+        await agents.register_agent(client, agent, steps=[])
+
+        control_ids: list[int] = []
+        for control_name, control_data in CONTROL_SPECS:
+            control_id = await _ensure_control(client, control_name, control_data)
+            control_ids.append(control_id)
+            print(f"Prepared control: {control_name} ({control_id})")
+
+        for control_id in control_ids:
+            try:
+                await agents.add_agent_control(client, AGENT_NAME, control_id)
+            except httpx.HTTPStatusError as exc:
+                if exc.response.status_code != 409:
+                    raise
+
+        print()
+        print("LangChain middleware smoke test is ready.")
+        print("Run: uv run langchain_middleware_smoke.py")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())