Quickstart

from typing import Annotated, TypedDict

from langchain_core.messages import AIMessage, AnyMessage
from langgraph.graph import END, START, StateGraph
from langgraph.graph.message import add_messages


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


def reply_node(state: State) -> dict:
    inbound = state["messages"][-1]
    return {"messages": [AIMessage(content=f"Echo: {inbound.content}")]}


def build_graph():
    graph = StateGraph(State)
    graph.add_node("reply", reply_node)
    graph.add_edge(START, "reply")
    graph.add_edge("reply", END)
    return graph

aion:
  agents:
    support:
      path: "./agent.py:build_graph"
      framework: "langgraph"
      name: "Support Agent"
      capabilities:
        streaming: true

# .env
AION_CLIENT_ID=your_client_id
AION_CLIENT_SECRET=your_client_secret

poetry run aion serve

# Server health check
curl http://localhost:10000/health/
# Server agent manifest
curl http://localhost:10000/.well-known/manifest.json
# A2A agent card for the "support" agent
curl http://localhost:10000/agents/support/.well-known/agent-card.json

poetry run aion chat

pip install aion-authoring-langgraph

from langgraph.graph import END, START, MessagesState, StateGraph
from langgraph.runtime import Runtime

from aion.core.runtime import AionRuntimeContext
from aion.langgraph.authoring import Thread


async def reply_node(state: MessagesState, runtime: Runtime[AionRuntimeContext]):
    thread = Thread.from_context(runtime.context)
    await thread.reply("I received that.")
    return {}


builder = StateGraph(MessagesState, context_schema=AionRuntimeContext)
builder.add_node("reply", reply_node)
builder.add_edge(START, "reply")
builder.add_edge("reply", END)
graph = builder.compile()