Metadata-Version: 2.5
Name: reflex-agentkit
Version: 0.2.0
Summary: The production UI layer for Python AI agents, built on Reflex and the AG-UI protocol.
Project-URL: Homepage, https://github.com/ecrespo/reflex-agentkit
Project-URL: Documentation, https://blog.seraph.to/reflex-agentkit/
Project-URL: Repository, https://github.com/ecrespo/reflex-agentkit
Project-URL: Issues, https://github.com/ecrespo/reflex-agentkit/issues
Project-URL: Changelog, https://github.com/ecrespo/reflex-agentkit/blob/main/CHANGELOG.md
Author-email: Ernesto Crespo <ecrespo@gmail.com>
License-Expression: MIT
License-File: LICENSE
Keywords: ag-ui,agents,chat,langgraph,llm,pydantic-ai,reflex
Classifier: Development Status :: 2 - Pre-Alpha
Classifier: Framework :: AsyncIO
Classifier: Intended Audience :: Developers
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Topic :: Scientific/Engineering :: Artificial Intelligence
Classifier: Typing :: Typed
Requires-Python: <3.14,>=3.11
Requires-Dist: ag-ui-protocol<2,>=1.0.0
Requires-Dist: httpx<1,>=0.28
Requires-Dist: jsonpatch<2,>=1.33
Requires-Dist: pydantic<3,>=2.11.2
Requires-Dist: reflex<0.11,>=0.10.1
Provides-Extra: all
Requires-Dist: ag-ui-langgraph[fastapi]<0.1,>=0.0.47; extra == 'all'
Requires-Dist: langgraph<2,>=1.2; extra == 'all'
Requires-Dist: pydantic-ai-slim[ag-ui,anthropic,openai]<3,>=2.64; extra == 'all'
Provides-Extra: langgraph
Requires-Dist: ag-ui-langgraph[fastapi]<0.1,>=0.0.47; extra == 'langgraph'
Requires-Dist: langgraph<2,>=1.2; extra == 'langgraph'
Provides-Extra: pydantic-ai
Requires-Dist: pydantic-ai-slim[ag-ui,anthropic,openai]<3,>=2.64; extra == 'pydantic-ai'
Description-Content-Type: text/markdown

# reflex-agentkit

Build and serve AI agents in pure Python with Reflex. AG-UI protocol client, streaming
chat, tool-call cards and human-in-the-loop approval for LangGraph, PydanticAI and any
AG-UI-compatible agent. The production UI layer your agent was missing — no TypeScript,
no Streamlit.

> **Status: `0.2.0`.** Everything in v0.1 is implemented and tested: the pure core,
> the agent clients, the Reflex state engine (`AgentState`), the chat components, both
> adapters (PydanticAI and LangGraph) and the AG-UI Dojo examples. `0.2.0` tells the model
> when the user edits a tool call's arguments before approving it.
> `pip install "reflex-agentkit[pydantic-ai]"` or `uv add "reflex-agentkit[pydantic-ai]"`. Documentation: https://blog.seraph.to/reflex-agentkit/

## Why

A Python developer with a working agent usually has two options for its UI: Streamlit,
which re-runs the whole script on every interaction and has to work around streaming tool
calls, mid-run approvals and cancellation; or a Next.js frontend, which adds a second
language, build and deployment. reflex-agentkit is the middle: Reflex already holds the
state on the server, keeps a WebSocket to every browser and runs background tasks, and
this library adds the agent layer on top, speaking the AG-UI protocol that PydanticAI and
LangGraph already emit.

## Quickstart

A streaming chat with a PydanticAI travel agent, one tool call shown as a card and one
tool that waits for your approval. This is the whole app
([`examples/quickstart/`](examples/quickstart/)):

<!-- quickstart:start -->
```python
from typing import ClassVar

import reflex as rx
from pydantic_ai import Agent

from reflex_agentkit import AgentState, agent_chat
from reflex_agentkit.adapters.pydantic_ai import PydanticAIAgent

travel_agent = Agent(
    "anthropic:claude-sonnet-5-5",
    instructions="You are a helpful travel assistant.",
)


@travel_agent.tool_plain
def get_weather(city: str) -> str:
    """Get the current weather for a city."""
    return f"It is sunny and 22 °C in {city}."


@travel_agent.tool_plain(requires_approval=True)
def book_hotel(city: str, nights: int) -> str:
    """Book a hotel room. Requires human approval."""
    return f"Booked {nights} nights in {city}."


class ChatState(AgentState, rx.State):
    agent: ClassVar = PydanticAIAgent(travel_agent)


def index() -> rx.Component:
    return agent_chat(ChatState)


app = rx.App()
app.add_page(index)
```
<!-- quickstart:end -->

Run it from a clone of this repo:

```bash
uv sync --all-extras
cd examples/quickstart
ANTHROPIC_API_KEY=sk-ant-... uv run --package quickstart reflex run
```

Ask it to plan a trip: it shows the `get_weather` call and opens the approval dialog
before `book_hotel` runs. The agentkit components are built on Radix Themes, so the
example's `rxconfig.py` adds `rx.plugins.RadixThemesPlugin()`.

To use a local model instead of Claude, swap the model line for an Ollama model through
PydanticAI's Ollama provider, for example `Agent("ollama:qwen3", ...)` with
`OLLAMA_BASE_URL=http://localhost:11434/v1`.

The components are in `reflex_agentkit.components`: `agent_chat` composes
`message_list`, `tool_call_card`, `composer`, `run_status_bar`, `approval_dialog` and,
with `show_shared_state=True`, `shared_state_panel`. Custom renderers for a tool
(`tool_renderers={"get_weather": ...}`) or an interrupt reason
(`interrupt_renderers={...}`) replace the default card body or form. Every control has
an accessible label.

## Features

| Feature | How |
|---|---|
| Streaming replies | Markdown streamed token by token, batched to at most one push per 50 ms |
| Tool calls | A card per call with arguments, status and result; custom renderers per tool |
| Approvals | Approve, edit the arguments, or reject a tool call before it runs |
| Questions from the agent | AG-UI standard interrupts, answered with a replaceable form |
| Shared state | The agent's state mirrored in `shared_state`, optionally validated by a Pydantic model, editable by the app |
| Client tools | Tools the agent calls that run in the Reflex app, e.g. to draw a chart |
| Real cancellation | Stop closes the agent's stream within a second, also across workers |
| Any AG-UI agent | In process (PydanticAI, LangGraph adapters) or over HTTP (`HttpAGUIClient`) |
| Safe by default | Agent, clients and keys never become browser state; every handler validates its input |

## How it compares

| | reflex-agentkit | Streamlit | CopilotKit | agentstage |
|---|---|---|---|---|
| UI written in | Python (Reflex) | Python | React / TypeScript | Python |
| Agent protocol | AG-UI | none (your own wiring) | AG-UI | its own JSON API over SSE |
| Agent frameworks | PydanticAI, LangGraph, any AG-UI server | any, by hand | many, through AG-UI | LangChain, LangGraph |
| Streaming and tool calls | built in | by hand, script reruns | built in | built in |
| Approvals and questions | AG-UI interrupts, edit before approving | by hand | built in | LangGraph interrupts |
| Shared state, client tools | built in | by hand | built in | — |
| Your own UI | any Reflex component | any Streamlit widget | any React component | configured app |

*agentstage (`agentstage-ui` on PyPI, 0.1.2, September 2026) per its README.*

## Documentation

The MkDocs site in [`docs/`](docs/) covers getting started (uv and pip), the concepts,
guides (your own UI, human in the loop, sidecar deployment, local models with Ollama), the
API reference, the security model and troubleshooting. It is published at
<https://blog.seraph.to/reflex-agentkit/> from `main`; preview it locally with
`uv run mkdocs serve`.

## Progress

Work follows the approved spec chain in [`specs/`](specs/) and the task list in
[`specs/tasks/agentkit-tasks.md`](specs/tasks/agentkit-tasks.md).

| Phase | Scope | Status |
|---|---|---|
| WP0 — Foundation | Packaging (uv, hatchling), lint, types, tests, CI | ✅ done |
| WP1 — Pure core | Models, reducer, commands, client tools, `RunInputBuilder`, property tests | ✅ done |
| WP2 — Clients | `AgentClient`/`AGUIAgent` protocols, SSE parser, `HttpAGUIClient`, `InProcessAGUIClient` | ✅ done |
| WP3 — Reflex state engine | `AgentState` mixin, streaming driver, cancellation, client-tool rounds | ✅ done |
| WP4 — PydanticAI adapter | `PydanticAIAgent`: a PydanticAI agent as an `AGUIAgent`, tool approvals, client tools | ✅ done |
| WP5 — Components and quickstart | `agent_chat` and its parts, quickstart example, release workflow | ✅ done |
| WP6 — LangGraph adapter | `LangGraphAGUIAgent`, `request_tool_approval`, legacy interrupts | ✅ done |
| WP7 — Dojo examples | The five AG-UI Dojo demos for PydanticAI and LangGraph | ✅ done |
| WP8 — Hardening, docs, release | Security review, benchmarks, documentation site, `0.1.0` on PyPI | ✅ done |
| Change: edited-args notice | REQ-035: the model learns about edited tool arguments (`changes/_archive/`) | ✅ `0.2.0` |

## What works today: `reflex_agentkit.core`

The core never imports `reflex`. AG-UI is its only event model, and a pure, total reducer
is the only way agent events change the conversation state.

- **`reduce(state, event)`**: handles all 31 AG-UI 1.0 event types (lifecycle and
  `RUN_FINISHED` outcomes, text, tool calls, shared state via JSON Patch, messages
  snapshots, reasoning, custom events). It never raises: problems become diagnostics.
- **Commands** (`reflex_agentkit.core.commands`): `begin_run`, `apply_approval`
  (optionally with edited arguments), `apply_rejection`, `apply_interrupt_response`,
  `apply_cancel`, `apply_client_tool_result`, `new_thread`, `resume_entries`.
  Human-in-the-loop follows AG-UI standard interrupts and the PydanticAI approval payload.
- **Client tools**: `@client_tool` builds the JSON Schema from type hints. `invoke()`
  runs sync handlers in a worker thread under a timeout and returns failures as tool
  results.
- **`RunInputBuilder`**: turns the conversation into an AG-UI `RunAgentInput` (full
  history, encrypted reasoning, edited tool args, shared state, tools, resume entries).
- **Agent clients**: `HttpAGUIClient` talks to any AG-UI server over HTTP + SSE.
  `InProcessAGUIClient` runs an agent object (anything with
  `run(run_input) -> AsyncIterator[event]`) in the same event loop. Both yield validated
  events, skip invalid ones with a log record, raise typed `AgentHTTPError` /
  `AgentStreamError` on transport failures, and close the upstream when the consumer is
  cancelled. Headers such as API keys are never logged or put in error messages.

```python
from reflex_agentkit.core import ConversationState, RunInputBuilder, client_tool
from reflex_agentkit.core.commands import apply_approval, begin_run
from reflex_agentkit.core.models import ReducerConfig
from reflex_agentkit.core.reducer import parse_event, reduce


@client_tool
def pick_date(day: int) -> str:
    """Let the user pick a day of the month."""
    return f"2026-05-{day:02d}"


config = ReducerConfig(client_tool_names=frozenset({"pick_date"}))
conv = begin_run(ConversationState.initial(config), config=config, user_text="book it")
run_input = RunInputBuilder(client_tools=(pick_date,)).build(conv)  # send to the agent

for raw in [  # AG-UI events as they arrive on the wire
    {"type": "RUN_STARTED", "threadId": conv.thread_id, "runId": conv.run_id},
    {"type": "TOOL_CALL_START", "toolCallId": "c1", "toolCallName": "book_hotel"},
    {"type": "TOOL_CALL_ARGS", "toolCallId": "c1", "delta": '{"city": "Lisbon"}'},
    {"type": "TOOL_CALL_END", "toolCallId": "c1"},
    {"type": "RUN_FINISHED", "threadId": conv.thread_id, "runId": conv.run_id,
     "outcome": {"type": "interrupt",
                 "interrupts": [{"id": "int-c1", "reason": "tool_call", "toolCallId": "c1"}]}},
]:
    event = parse_event(raw)
    conv = reduce(conv, event)

print(conv.run_status)                        # awaiting_input
conv = apply_approval(conv, "c1", {"city": "Porto"})
conv = begin_run(conv, config=config, resume=True)
print(RunInputBuilder().build(conv).resume)   # [ResumeEntry(..., payload={'approved': True, 'editedArgs': {...}})]
```

Streaming a run from an AG-UI server:

```python
import asyncio
import os

from reflex_agentkit import HttpAGUIClient
from reflex_agentkit.core import ConversationState, RunInputBuilder
from reflex_agentkit.core.commands import begin_run
from reflex_agentkit.core.models import ReducerConfig
from reflex_agentkit.core.reducer import reduce


async def main() -> None:
    client = HttpAGUIClient(
        "http://localhost:8000/agent",
        headers=lambda: {"Authorization": f"Bearer {os.environ['AGENT_TOKEN']}"},
    )
    config = ReducerConfig()
    conv = begin_run(ConversationState.initial(config), config=config, user_text="Hello!")
    try:
        async for event in client.stream(RunInputBuilder().build(conv)):
            conv = reduce(conv, event)
    finally:
        await client.aclose()
    print(conv.run_status, conv.messages[-1].content)


asyncio.run(main())
```

## `AgentState` with your own UI

`AgentState` is a Reflex state mixin that holds a conversation with an AG-UI agent. Its
configuration is class-level (`ClassVar`), so the agent, its client and any API keys
never become state vars sent to the browser. `agent_chat` is optional: this example
builds its own UI from plain Reflex components:

```python
from typing import ClassVar

import reflex as rx

from reflex_agentkit import AgentState, HttpAGUIClient


class ChatState(AgentState, rx.State):
    agent: ClassVar = HttpAGUIClient("http://localhost:8000/agent")
    draft: str = ""

    @rx.event
    def set_draft(self, value: str) -> None:
        self.draft = value

    @rx.event
    def submit(self):
        text, self.draft = self.draft, ""
        return ChatState.send_message(text)


def index() -> rx.Component:
    return rx.vstack(
        rx.foreach(ChatState.messages, lambda m: rx.text(m.role, ": ", m.content)),
        rx.cond(ChatState.error != "", rx.callout(ChatState.error, color_scheme="red")),
        rx.hstack(
            rx.input(value=ChatState.draft, on_change=ChatState.set_draft),
            rx.button("Send", on_click=ChatState.submit, disabled=~ChatState.can_send),
            rx.button("Stop", on_click=ChatState.cancel_run, disabled=~ChatState.is_running),
        ),
    )


app = rx.App()
app.add_page(index)
```

- **Streaming**: the reply streams in a background task; state pushes are batched (at most
  one per `flush_interval_ms`, default 50 ms).
- **Handlers**: `send_message`, `cancel_run` (stops within 1 s, also when another backend
  worker handles the click), `approve_tool` / `reject_tool` (with edited arguments),
  `respond_to_interrupt`, `new_thread`, `set_shared_state` / `patch_shared_state`,
  `retry_run`. Calls that are not allowed in the current status are no-ops, so they are
  safe to call from the browser at any time.
- **Client tools**: `client_tools: ClassVar = (my_tool,)` runs `@client_tool` handlers on
  the server and starts the follow-up run, with optional approval
  (`@client_tool(requires_approval=True)`).
- **Hooks**: override `create_client` (for example, per-user auth headers),
  `prepare_run_input`, `on_custom_event` or `on_run_end`.

## What works today: the PydanticAI adapter

Install the extra (`reflex-agentkit[pydantic-ai]`) and wrap the agent. Tools declared with
`requires_approval=True` pause the run for `approve_tool` / `reject_tool` with no extra
configuration; edited arguments reach the tool, and the tool's result tells the model the
user changed them (`notify_edited_args=False` turns that off).

```python
from typing import ClassVar

import reflex as rx
from pydantic_ai import Agent

from reflex_agentkit import AgentState
from reflex_agentkit.adapters.pydantic_ai import PydanticAIAgent

agent = Agent("anthropic:claude-sonnet-5-5", instructions="You book hotels.")


@agent.tool_plain(requires_approval=True)
def book_hotel(city: str) -> str:
    """Book a hotel in a city."""
    return f"Booked a hotel in {city}."


class ChatState(AgentState, rx.State):
    agent: ClassVar = PydanticAIAgent(agent)
```

## What works today: the LangGraph adapter

Install the extra (`reflex-agentkit[langgraph]`) and wrap a compiled graph that has a
checkpointer. `interrupt(value)` in the graph shows up as `pending_interrupt` and is
answered with `respond_to_interrupt`. `request_tool_approval` pauses for a tool call and
shows the approval dialog; it returns the decision, with edited arguments when the user
changed them. In this sketch, `call_model` stands for your own chat-model node:

```python
from typing import Any, ClassVar

import reflex as rx
from langchain_core.messages import AIMessage, ToolMessage
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import END, START, MessagesState, StateGraph

from reflex_agentkit import AgentState, agent_chat
from reflex_agentkit.adapters.langgraph import LangGraphAGUIAgent, request_tool_approval


def tools(state: MessagesState) -> dict[str, Any]:
    last = state["messages"][-1]
    assert isinstance(last, AIMessage)
    call = last.tool_calls[0]
    decision = request_tool_approval(str(call["id"]), call["name"], call["args"])
    content = f"Booked {decision.args['city']}" if decision.approved else "Not booked"
    content = decision.tool_result(content)  # tells the model about edited arguments
    return {"messages": [ToolMessage(content=content, tool_call_id=call["id"])]}


def route_to_tools_or_end(state: MessagesState) -> str:
    return "tools" if getattr(state["messages"][-1], "tool_calls", None) else END


builder = StateGraph(MessagesState)
builder.add_node("model", call_model)  # your chat-model node, e.g. ChatAnthropic + bind_tools
builder.add_node("tools", tools)
builder.add_edge(START, "model")
builder.add_conditional_edges("model", route_to_tools_or_end, ["tools", END])
builder.add_edge("tools", "model")


class ChatState(AgentState, rx.State):
    agent: ClassVar = LangGraphAGUIAgent(builder.compile(checkpointer=InMemorySaver()))


def index() -> rx.Component:
    return agent_chat(ChatState)
```

- One agent instance serves every session; each run works on a fresh copy.
- Several interrupts at once are answered one at a time in the UI and resumed together,
  each node receiving its own answer.
- Cancelling while paused resumes the next run with
  `ApprovalDecision(approved=False, reason="cancelled")` for an approval, or the
  `{"__agui_cancelled__": True}` value for a generic interrupt.
- `legacy_interrupts=True` supports integrations that only send the older `on_interrupt`
  event.

## Examples

| Example | What it is |
|---|---|
| [`examples/quickstart/`](examples/quickstart/) | The quickstart above: a chat with a PydanticAI agent in 23 lines |
| [`examples/pydantic_ai_dojo/`](examples/pydantic_ai_dojo/) | The five AG-UI Dojo demos with PydanticAI |
| [`examples/langgraph_dojo/`](examples/langgraph_dojo/) | The same five demos with LangGraph graphs |

The Dojo demos are agentic chat, backend tool rendering, human in the loop (approve, edit
or reject a tool call; answer a question), shared state and tool-based generative UI.
They use Claude by default; `AGENTKIT_MODEL=ollama:<model>` switches to a local Ollama
model, and `AGENTKIT_MODEL=test` to an offline model (see
[`examples/_shared`](examples/_shared/README.md)).

## Development

The workflow is uv only (Python 3.11–3.13).

| Purpose | Command |
|---|---|
| Set up the environment | `uv sync --all-extras --all-packages` |
| Lint and format check | `uv run ruff check .` · `uv run ruff format --check .` |
| Type check (strict on `core` and `state`) | `uv run mypy` |
| Tests with coverage | `uv run pytest` |
| Coverage gate (≥ 85 % on `core` and `state`) | `uv run python scripts/check_gated_coverage.py` |
| Core isolation check | `uv run python scripts/assert_core_isolated.py` |
| Quickstart smoke (offline) | `uv run pytest tests/smoke` |
| Benchmarks (NFR-01, NFR-04) | `uv run pytest -m bench -q` |
| Docs site | `uv run mkdocs serve` · `uv run mkdocs build --strict` |

Releases are automatic: merging a new version into `main` publishes it to PyPI (see
[`CONTRIBUTING.md`](CONTRIBUTING.md#releases)).

Read [`CONTRIBUTING.md`](CONTRIBUTING.md) and [`specs/constitution.md`](specs/constitution.md)
before contributing. The rules: the specs come first, `core` and `state` are written
test-first, and changes use Conventional Commits.

## License

MIT
