LangGraph

from typing import Annotated, TypedDict

from langchain_core.messages import AnyMessage, SystemMessage, ToolMessage
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI
from langgraph.graph import END, START, StateGraph
from langgraph.graph.message import add_messages
from langgraph.types import interrupt

import agentos
from agentos.adapters import langgraph

ORDERS = {"A-1001": {"status": "delivered", "total": "49.90", "currency": "EUR"}}


@tool
def lookup_order(order_id: str) -> dict:
    """Look an order up by its id, e.g. A-1001."""
    return ORDERS.get(order_id, {"error": f"no order {order_id}"})


@tool
def refund(order_id: str, amount: str) -> str:
    """Refund an order. The amount is text, e.g. "49.90". A person approves every refund first."""
    return f"refunded {amount} for order {order_id}"


TOOLS = {t.name: t for t in (lookup_order, refund)}
NEEDS_APPROVAL = {"refund"}


class State(TypedDict):
    messages: Annotated[list[AnyMessage], add_messages]


def chat_model() -> ChatOpenAI:
    """config.models.main through the gateway, with this invocation's run token."""
    inv = agentos.current()
    return ChatOpenAI(model=inv.llm.model(), base_url=inv.llm.base_url, api_key=inv.llm.api_key,
                      default_headers=inv.llm.headers, http_client=inv.llm.httpx_client(),
                      http_async_client=inv.llm.httpx_client(asynchronous=True))


def agent(state: State) -> State:
    system = SystemMessage(agentos.current().config["prompts"]["system"])
    reply = chat_model().bind_tools(list(TOOLS.values())).invoke([system, *state["messages"]])
    return {"messages": [reply]}


def run_tools(state: State) -> State:
    calls = state["messages"][-1].tool_calls
    asked = [call for call in calls if call["name"] in NEEDS_APPROVAL]
    decisions = {}
    if asked:  # one hold for every call that needs a person; the run waits here
        answer = interrupt({
            "intent": "Approve a refund",
            "actions": ["approve", "edit", "reject"],
            "items": [{"id": call["id"], "title": f"Refund {call['args'].get('amount')} for order "
                                                  f"{call['args'].get('order_id')}", "data": call["args"]}
                      for call in asked],
        })
        decisions = {item["id"]: item for item in answer["items"]}
    results = []
    for call in calls:
        decision = decisions.get(call["id"])
        if decision and decision["action"] not in ("approve", "edit"):
            content = decision["comment"] or "A person rejected this refund."
        else:
            content = TOOLS[call["name"]].invoke(decision["data"] if decision else call["args"])
        results.append(ToolMessage(content=str(content), tool_call_id=call["id"]))
    return {"messages": results}


def next_step(state: State) -> str:
    return "tools" if state["messages"][-1].tool_calls else END


builder = StateGraph(State)
builder.add_node("agent", agent)
builder.add_node("tools", run_tools)
builder.add_edge(START, "agent")
builder.add_conditional_edges("agent", next_step, ["tools", END])
builder.add_edge("tools", "agent")


def support(inv: agentos.Invocation):
    graph = builder.compile(checkpointer=langgraph.checkpointer())  # per invocation: checkpoints live in run state
    try:
        state = langgraph.run(inv, graph, {"messages": [("user", inv.input["message"])]})
    except Exception as error:
        if block := agentos.guardrail_block(error):  # a guardrail refused a model call
            raise block from error
        raise
    return {"answer": state["messages"][-1].content}


if __name__ == "__main__":
    agentos.serve({"support": support}, port=8080)

name: support-langgraph
title: Support agent (LangGraph)
description: Answers order questions with a lookup tool; refunds go through a person first.
runtime: stateless
port: 8080
capabilities:
  support:
    description: Handles a customer's message. A refund the model wants to issue waits for a person's approval.
    input:
      type: object
      properties: {message: {type: string, minLength: 1}}
      required: [message]
    output:
      type: object
      properties: {answer: {type: string}}
      required: [answer]
    timeout: 120s
config:
  models:
    main: anthropic/claude-sonnet-4-5    # a model from your Models page: <provider slug>/<model ID>
  prompts:
    system: >-
      You are a support agent for an online shop. Look orders up with lookup_order before you
      answer anything about an order. Refund only with the refund tool. Never refund more than the order's total.

FROM python:3.13-slim
WORKDIR /app
COPY vendor/ vendor/
COPY requirements.txt .
RUN pip install --no-cache-dir "$(ls vendor/*.whl)[langgraph]" -r requirements.txt
COPY app.py .
USER nobody
EXPOSE 8080
CMD ["python", "-u", "app.py"]

langgraph==1.2.15
langchain-openai==1.7.3