langstage-core

The shared core behind the LangStage family: a host layer for LangGraph agents (spec-loading + layered config), an in-process AG-UI bridge that streams any CompiledGraph to a frontend, an async task-delegation engine, and interrupt-aware input helpers. Write your agent once — any LangGraph CompiledGraph — and every LangStage surface runs it the same way.

1.0 — renamed from langgraph-stream-parser. The old StreamParser / events / event_to_dict event layer was retired in favor of the AG-UI wire (see Migrating and ADR 0003). The old import langgraph_stream_parser keeps working as long as the separate langgraph-stream-parser compat package stays installed (it re-exports langstage_core); a fresh install of langstage-core alone does not provide it.

Every stage for your LangGraph agent

langstage-core is the shared core of the LangStage family: write your agent once — any LangGraph CompiledGraph — and run it on every stage with the same spec string (module:attr or path/to/file.py:attr), the same langstage.toml config file, and the same LANGSTAGE_* environment variables. (The pre-rename deepagents.toml / DEEPAGENT_* vocabulary still resolves as a deprecated fallback.)

📖 Full documentation: https://dkedar7.github.io/langstage-docs/

Installation

pip install "langstage-core[agui]"

The [agui] extra pulls the AG-UI runtime (ag-ui-langgraph[fastapi] + uvicorn) — needed for the streaming bridge below and by every LangStage surface. The bare pip install langstage-core (only langchain-core) is enough if you just want the host/config/tasks layer without streaming.

No agent of your own yet? The [stub] extra adds a keyless echo graph you can stream:

pip install "langstage-core[agui,stub]"

Quick start

Wrap any compiled graph with build_agent, then stream a turn. Two shared mappings cover the two frontend styles the family uses:

import asyncio
from langstage_core import load_agent_spec
from langstage_core.agui import build_agent, iter_event_frames

# any LangGraph CompiledGraph — here the keyless demo stub
agent = build_agent(load_agent_spec("langstage_core.demo.stub:graph"))

async def main():
    async for frame in iter_event_frames(agent, "hello", thread_id="s1"):
        if frame["type"] == "content":
            print(frame["content"], end="")
        elif frame["type"] == "complete":
            print()

asyncio.run(main())

• iter_event_frames yields rich, typed frames — content, tool_start, tool_end, reasoning, interrupt, extraction, complete, error — used by the web and VS Code surfaces.
• iter_chunk_frames yields terminal-friendly chunk dicts — {"status": "streaming", "chunk": "..."} … {"status": "complete"} — used by the CLI and Jupyter surfaces.

build_agent attaches an in-memory checkpointer if the graph has none, so multi-turn memory and interrupts work out of the box; pass a thread_id per turn to key per-conversation state.

See every frame type, keyless

The echo stub above only emits content. To see the rich frames without an API key, point build_agent at the bundled tool demo (langstage_core.demo.tools:graph): it calls a built-in tool through a real ToolNode, streams a reasoning delta, and raises a resumable interrupt, all deterministically and offline. Each trigger phrase drives a different frame type:

import asyncio
from langstage_core import create_resume_input
from langstage_core.agui import build_agent, iter_event_frames
from langstage_core.demo.tools import create_tool_demo_agent, demo_extractors

agent = build_agent(create_tool_demo_agent())

async def main():
    for turn in ("hello", "think about it", "use a tool"):
        async for frame in iter_event_frames(agent, turn, "s1", extractors=demo_extractors()):
            print(frame["type"], "→", {k: v for k, v in frame.items() if k != "type"})

    # "ask me" raises interrupt(...); resume the same thread with a decision.
    async for frame in iter_event_frames(agent, "ask me", "s2"):
        print(frame["type"])                       # ... interrupt
    async for frame in iter_event_frames(agent, "", "s2",
                                         resume=create_resume_input(decisions=[{"type": "approve"}])):
        print(frame["type"])                       # content, complete

asyncio.run(main())
# content · reasoning · tool_start · tool_end · extraction · interrupt · complete

Serve the same demo over AG-UI with langstage-agui --demo=tools.

One call, one answer (no streaming)

The iter_* mappings are streaming generators — perfect for a live UI, but a test, an eval/grading harness, a batch job, or a "run my agent once, give me the answer" script wants a single call that returns the result. run_turn (sync) / collect_event_frames (async) do exactly that, returning a typed TurnResult (text, tool_calls, extractions, reasoning, outcome, interrupt, error, frames) — nothing streamed, nothing hand-accumulated:

from langstage_core.agui import run_turn
from langstage_core.demo.tools import create_tool_demo_agent, demo_extractors

result = run_turn(create_tool_demo_agent(), "use a tool", extractors=demo_extractors())
result.text          # 'The demo tool returned {"query": "use a tool", "answer": "42", ...}'
result.tool_calls    # [{'name': 'demo_lookup', 'args': {'query': 'use a tool'}, 'id': 'demo_lookup_1'}]
result.extractions   # [{'tool_name': 'demo_lookup', 'extracted_type': 'demo_fact', 'data': {...}}]
result.outcome       # 'complete'   ('interrupted' on "ask me", 'error' on a failing turn)

run_turn accepts a compiled graph or a prebuilt build_agent(...) and runs the turn under asyncio.run; inside an event loop, await collect_event_frames(agent, message, thread_id, ...) instead (or collect_chunk_frames for the chunk wire). The complete / interrupted / error verdict is the same rule SessionAdapter uses, so a one-shot turn and a streamed one agree. (The sibling langstage package's oneturn.py is a different layer — it buffers a SessionAdapter for the web one-turn HTTP endpoint; these core helpers are session-free, for tests/evals/scripts.)

Connect a real model

The demos above are keyless. To stream your own model-backed agent, bring any LangGraph CompiledGraph — nothing about the library is demo-specific. The [real] extra pulls a lightweight OpenAI-compatible stack:

pip install "langstage-core[agui,real]"   # langchain-openai + langgraph

import asyncio, os
from langchain_openai import ChatOpenAI
from langgraph.prebuilt import create_react_agent
from langstage_core.agui import build_agent, iter_event_frames

# Works with OpenAI, OpenRouter, or any OpenAI-compatible endpoint:
model = ChatOpenAI(
    model="gpt-4o-mini",
    base_url=os.environ.get("OPENAI_BASE_URL"),   # e.g. https://openrouter.ai/api/v1
    api_key=os.environ["OPENAI_API_KEY"],
)
agent = build_agent(create_react_agent(model, tools=[]))

async def main():
    async for frame in iter_event_frames(agent, "Say hi in one word.", thread_id="s1"):
        if frame["type"] == "content":
            print(frame["content"], end="")

asyncio.run(main())

Everything else — run_turn, serve, the task engine, extractors — takes the same build_agent(...) agent, so the keyless snippets above work verbatim against a real model once you swap the graph. (Prefer Anthropic + the full agent stack? pip install deepagents langchain-anthropic and build a deepagents graph instead; the library only ever sees a CompiledGraph.)

Delegate work to a background task

The task engine is a single-process worker pool: enqueue a prompt, walk away, and read the result off the board when it's done. Any CompiledGraph drives the workers.

import asyncio
from langstage_core import SessionAdapter, load_agent_spec
from langstage_core.tasks import TaskRunner, InMemoryTaskStore, TERMINAL_STATES

async def main():
    adapter = SessionAdapter(graph=load_agent_spec("langstage_core.demo.stub:graph"))
    runner = TaskRunner(adapter, InMemoryTaskStore(), concurrency=3)
    await runner.start()

    task_id = await runner.enqueue(title="research", prompt="Summarize the plan.")

    # delegate-and-walk-away: poll the board until the task reaches a terminal state
    while (task := await runner.store.get(task_id))["state"] not in TERMINAL_STATES:
        await asyncio.sleep(0.1)

    print(task["state"])    # 'done'
    print(task["result"])   # the agent's answer
    await runner.shutdown()

asyncio.run(main())

A Task is a TypedDict — read it with task["state"] / task["result"] / task["error"] / task["interrupt"], not attribute access. States flow queued → ongoing → review_needed → done | failed | cancelled; TERMINAL_STATES is the set to stop polling on. TASK_TOOLS (with set_runner / get_runner) are the agent-facing delegation tools, so an agent can enqueue background work to copies of itself.

Human-in-the-loop (interrupt → resume)

When the graph calls interrupt(...), you get an interrupt frame; resume by passing the decision back via resume=:

async for frame in iter_event_frames(agent, "run it", thread_id="s1"):
    if frame["type"] == "interrupt":
        # frame["action_requests"], frame["allowed_decisions"]
        ...

# next turn resumes the same thread with the user's decision
async for frame in iter_event_frames(agent, "", thread_id="s1",
                                     resume={"decisions": [{"type": "approve"}]}):
    ...

Decision types: approve, reject, edit, respond (deepagents 0.6+ / LangGraph 1.1+).

What's in the box

Everything is re-exported from the top-level langstage_core package (except the AG-UI helpers under langstage_core.agui):

Serve any agent over AG-UI

Any LangGraph agent can be served over the AG-UI protocol — the event-based wire for streaming rich agent interactions (text, tool calls, reasoning, state, interrupts) to frontends (CopilotKit, React/Vue/Angular components, any AG-UI client). The host layer resolves which agent; the official MIT ag-ui-langgraph adapter owns the wire:

langstage-agui --agent my_agent.py:graph     # serve over AG-UI at http://localhost:8050
langstage-agui --demo                          # keyless echo agent, no API key
langstage-agui --demo=tools                    # keyless rich-frame demo (tools, reasoning, interrupt)
langstage-agui --agent my_agent.py:graph --verify        # run one keyless turn; exit 0 ok / 1 failed
langstage-agui --agent my_agent.py:graph -m "hi there"   # run ONE turn with your prompt, print the reply

--verify is the preflight to run right after wiring up an agent: --show-config proves the config chain resolves a spec, but --verify proves it loads and actually produces a turn — catching the two most common failures (a typo'd module:attr, or a graph that loads but yields an empty/erroring turn) that otherwise only surface at first chat. Keyless, so it fits a CI/deploy gate. --message/-m is its companion — run one turn with your prompt and print the answer (add --json for the typed TurnResult), exit 0/1/2 on complete/error/interrupt. The three questions every adopter asks, in order: --show-config (resolves?) → --verify (runs?) → --message (what does it say?).

from langstage_core.agui import build_app
app = build_app(my_compiled_graph)   # an ASGI (FastAPI) app; run with uvicorn

See ADR 0001 for the rationale.

Configuration

The same resolution chain everywhere — defaults < langstage.toml < LANGSTAGE_* env < CLI/overrides (legacy deepagents.toml / DEEPAGENT_* still resolve as a deprecated fallback). Print the resolved value + source of every key:

python -m langstage_core.host      # or each surface's --show-config

Migrating from langgraph-stream-parser

langstage-core 1.0 is the rename of langgraph-stream-parser. The old import name keeps working through a separate compat package — langgraph-stream-parser 1.0, which now just re-exports langstage_core (with a DeprecationWarning). So import langgraph_stream_parser and its submodules keep resolving only while that package remains installed:

• Upgrading in place (pip install -U langgraph-stream-parser) → you keep the shim package, so the old import keeps working. Update to import langstage_core when convenient.
• Installing langstage-core fresh does not pull the shim (it's a separate distribution, and depending on it would be circular). Either import langstage_core (recommended), or pip install langgraph-stream-parser alongside if you need the old name during a transition.

The event layer was removed in 1.0. If you used it directly, migrate:

Kept and unchanged: load_agent_spec, HostConfig, prepare_agent_input, create_resume_input, the tasks engine, and the extractors (ToolExtractor + built-ins). Full detail: ADR 0003.

Development

pip install -e ".[dev]"
pytest
pytest --cov=langstage_core

License

MIT