Build your first agent

pip install aegra-cli
aegra init

cd <your-project>
cp .env.example .env

my-agent/
├── aegra.json              # Registers your graph
├── src/
│   └── my_agent/
│       ├── graph.py        # Agent logic and graph definition
│       ├── state.py        # Input and internal state schemas
│       ├── context.py      # Configurable parameters (model, prompts)
│       ├── tools.py        # Tools the agent can use
│       ├── prompts.py      # System prompt template
│       └── utils.py        # Model loading helper
├── .env.example
├── pyproject.toml
├── Dockerfile
└── docker-compose.yml

{
  "graphs": {
    "agent": "./src/my_agent/graph.py:graph"
  }
}

uv sync
uv run aegra dev

INFO:     Application startup complete.
INFO:     Uvicorn running on http://127.0.0.1:2026

import asyncio
from langgraph_sdk import get_client


async def main():
    client = get_client(url="http://localhost:2026")

    # Create an assistant from your graph
    assistant = await client.assistants.create(
        graph_id="agent",
        name="My Search Agent",
        metadata={"purpose": "web search"},
    )
    print(f"Assistant ID: {assistant['assistant_id']}")


asyncio.run(main())

import asyncio
from langgraph_sdk import get_client


async def main():
    client = get_client(url="http://localhost:2026")

    # Create a thread
    thread = await client.threads.create()
    print(f"Thread ID: {thread['thread_id']}")

    # Run the agent with streaming
    async for chunk in client.runs.stream(
        thread_id=thread["thread_id"],
        assistant_id="agent",  # graph ID works as assistant ID
        input={"messages": [{"type": "human", "content": "What is Aegra?"}]},
        stream_mode=["messages-tuple"],
    ):
        if hasattr(chunk, "data") and chunk.data:
            print(chunk.data)


asyncio.run(main())

async def continue_conversation():
    client = get_client(url="http://localhost:2026")

    thread_id = "your-thread-id-from-step-5"

    # Follow-up — the agent remembers the previous exchange
    async for chunk in client.runs.stream(
        thread_id=thread_id,
        assistant_id="agent",
        input={"messages": [{"type": "human", "content": "Tell me more about its features"}]},
        stream_mode=["messages-tuple"],
    ):
        if hasattr(chunk, "data") and chunk.data:
            print(chunk.data)


asyncio.run(continue_conversation())

async def inspect_state():
    client = get_client(url="http://localhost:2026")

    thread_id = "your-thread-id"

    # Get current state
    state = await client.threads.get_state(thread_id)
    print("Current values:", state["values"])
    print("Next nodes:", state["next"])

    # Get history (all checkpoints)
    history = await client.threads.get_history(thread_id)
    print(f"Total checkpoints: {len(history)}")
    for entry in history:
        print(f"  Checkpoint: {entry['checkpoint']['checkpoint_id']}")


asyncio.run(inspect_state())

git clone https://github.com/langchain-ai/agent-chat-ui.git
cd agent-chat-ui
pnpm install
pnpm dev