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 status-keyed chunk dicts — used by the CLI and Jupyter surfaces. A streaming chunk carries exactly one payload key (chunk, reasoning, tool_calls, tool_result, or extraction), so branch on the key rather than assuming chunk.

build_agent attaches an in-memory checkpointer if the graph has none (on a copy — your graph object is never mutated), so multi-turn memory and interrupts work out of the box: build the agent once, reuse it, and pass a thread_id per turn to key per-conversation state.

Frame reference

Both wires carry the same information; the table is the contract (keys marked new are additive and safe to ignore).

Frames arrive in message order: a node that returns a finished AIMessage (no token streaming — model.invoke(), a router, a canned reply) is emitted when that node finishes, its text before its own tool calls, and the served AG-UI endpoint (build_app / serve) streams the same TEXT_MESSAGE_* / TOOL_CALL_* events the in-process wires are built from.

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, complete (outcome="interrupted")
    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, and frames — an int frame count, not the frame list) — 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. Each call is an isolated one-shot by default — a fresh thread_id per call, and the graph you pass is not mutated — so for prompt in dataset: run_turn(graph, prompt) never leaks one turn into the next; to carry state across calls (or resume an interrupt), pass the same build_agent(...) agent and an explicit thread_id each time; 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+).

frame["allowed_decisions"] is the interrupt's own decision set, not a fixed list: a standard HumanInterrupt's config (allow_accept → approve, allow_edit → edit, allow_respond → respond, allow_ignore → reject) or a HumanInTheLoopMiddleware payload's per-action review_configs[*].allowed_decisions decide it, so an approve-only interrupt advertises exactly ["approve"]. The full four are the fallback only when the interrupt says nothing.

resume= takes the raw payload or a create_resume_input(...) Command. On ag-ui-langgraph ≥ 0.0.43 it is sent on the adapter's standard RunAgentInput.resume[] (answering the thread's pending interrupt), so a resume logs no forwardedProps.command.resume is deprecated / failed to parse … resume_input as JSON warning; older adapters, or a thread with several pending interrupts, keep the legacy forwarded_props.command.resume wire.

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?).

Every can't-run failure is a clean one-line error: on stderr, never a traceback (set LANGSTAGE_DEBUG=1 for one): an agent that loads but isn't a runnable graph (e.g. a StateGraph you forgot to .compile()) exits 1 under both --verify and --message (--json still prints a typed TurnResult with outcome: "error"). When serving, the port is bound before the Serving … at <url> banner prints, so a port already in use is error: cannot serve at <url>: … and exit 2 (the serve path's can't-start code, like an unloadable spec), not a success banner followed by a crash. serve() binds first too and raises OSError for a busy port.

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

What the diagnostic tells you:

• Every contributing file. TOML read from: lists the global ~/.langstage/config.toml and the project langstage.toml; config_dict()["toml"]["paths"] is the same list as data.
• A malformed file is reported as malformed, not missing. A langstage.toml that doesn't parse is ignored entirely (every key falls back to env/defaults) and shows as TOML: <path> is MALFORMED and was ignored entirely (<parse error>). In config_dict() it appears as toml.found: true, toml.malformed: true, and toml.malformed_files: [{path, error}].
• Anything ignored or degraded, as data. HostConfig.config_issues() (and config_dict()["issues"]) lists each malformed file, each wrong-type or invalid value that fell back to a default, and each unknown key. An empty list means the config is clean, so a surface's --strict gate can fail when the list isn't empty.
• debug is a top-level key. In TOML, debug = true must come before the first [table] header. Written below [server], TOML reads it as server.debug. That key is ignored, and a note: saying so is printed at startup.
• Booleans accept true/false, 0/1, and the same quoted strings as env vars ("yes", "off", ...). An unrecognized value falls back to the default and prints a note:.
• [configurable] keys are passed to the graph's config["configurable"] by langstage-agui (for both serving and --message), and --show-config lists them. thread_id is always set per run. Python callers pass build_agent(config=...) themselves.
• Legacy names (DEEPAGENT_*, DEEPAGENTS_CONFIG_HOME, deepagents.toml) each print exactly one note: per process. Set LANGSTAGE_SUPPRESS_LEGACY_NOTICE=1 to silence them.

Surfaces print user-controlled values, so they should print through langstage_core.console.safe_print / safe_write. These escape characters the console can't encode (a cp1252 Windows console, for example) instead of raising UnicodeEncodeError.

Agent specs and relative paths

A spec is path/to/file.py:attr or package.module:attr. The :attr suffix is required: a colon-less spec is an error, never a silent fallback to a default agent. Surrounding whitespace is ignored and a leading ~ is expanded. The attribute must be the agent object itself: a str attribute is rejected, not followed as another spec. load_agent_spec imports like python my_agent.py does:

• file.py:attr puts the file's own directory first on sys.path, so the agent can import its sibling modules (from tools import ...).
• package.module:attr falls back to the current directory (or base_dir=) when the package isn't otherwise importable. That covers a project-local package run from a console script.
• load_agent_spec(spec, stdout_to_stderr=True) sends the agent's import-time prints to stderr. Use it on machine-readable paths. langstage-agui --verify / -m / --json already do.

Relative paths in a TOML file resolve against that file's directory, like paths in pyproject.toml. This applies to [agent] spec (file form) and [workspace] root, for both the project langstage.toml (found by walking up from the cwd) and the global ~/.langstage/config.toml. A project therefore runs the same from its root and from any subdirectory. In the global file, relative paths resolve against ~/.langstage/, so write ~/agents/my_agent.py:graph or an absolute path there. Values from LANGSTAGE_* env vars and CLI flags stay relative to the cwd. For a dotted spec from TOML, cfg.toml_dir_for("agent_spec") gives you the file's directory to pass as base_dir=.

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