langchain-agentx-stream-ui-backend

SSE 薄层 SDK：仅 UI 线 — 将 LangchainAgentEvent（经 SDK LangGraphToLangchainAgentEventAdapter 的 out-of-band / artifact 投影）序列化为前端 SSE 协议。

不暴露 LangGraph 原始 dict，也不向浏览器推送 ToolMessage.content（model 路径，仅供 LLM）。

快速接入

from fastapi import FastAPI, Request
from langchain_agentx_stream_ui_backend import stream_to_sse, wrap_langgraph

app = FastAPI()

@app.get("/wiki/build")
async def wiki_build(task_id: str, request: Request):
    graph = build_wiki_workflow(task_id)
    # adapter_config 可省略：wrap_langgraph 内置 DEFAULT_UI_ADAPTER_CONFIG
    events = wrap_langgraph(graph, {"task_id": task_id}, graph_name="WikiBuilder")
    return stream_to_sse(events, request=request, session_id=f"wiki-{task_id}")

测试组织（对齐 T10 金字塔）

层级	位置	说明
L0 单测	`langchain_agentx_stream_ui_backend/test_.py`、`tests/test_.py`	函数/类级行为
L1 集成	`tests/integration/`	FastAPI 路由 + SSE / buffer / permission 全链路
冒烟	`tests/smoke/`	CI 快速回归（`-m smoke`，约 7 项）

原 examples/ 目录已合并为 tests/integration/sse/ 下的集成测试。

cd backend
python -m pip install -e ".[dev]"

# 单元测试（L0 + smoke + tests/test_*.py；默认 pytest 即此模式）
pytest -v
# 或
bash scripts/test-unit.sh

# 集成测试（L1；须单独跑，不与单测同进程）
bash scripts/test-integration.sh
# 或
pytest -o addopts= -m integration tests/integration/ -v

# 全量（串行两段，避免同进程加载 L0+L1 触发 teardown 问题）
bash scripts/test-unit.sh && bash scripts/test-integration.sh

# 冒烟（发布前 / CI 快速门禁）
pytest tests/smoke/ -v -m smoke

集成测试覆盖

模块	文件	覆盖点
Wiki / LangGraph	`test_wiki_workflow_integration.py`	`wrap_langgraph` + SSE
AgentSession	`test_agent_session_integration.py`	`wrap_agent_session` + SSE
EventBuffer	`test_buffer_multi_subscriber_integration.py`	多订阅、断连不解 ingest
Last-Event-ID	`test_last_event_id_integration.py`	SSE `id:` 续传、400 校验
Middleware	`test_middleware_http_integration.py`	HTTP 路由级 middleware
Permission raise	`test_permission_raise_integration.py`	ingest 挂起 → POST 回写 → 继续
协议不变量	`test_sse_protocol_integration.py`	meta 字段、bypass/raise 模式
断连	`test_stream_disconnect_integration.py`	`aclose` 2 秒内触发

公开 API

stream_to_sse(events, *, request, session_id, ...) → StreamingResponse（SseStreamingService 门面）
stream_to_sse(..., assign_event_ids=True) — 形态 A 单连接内 SSE id 行（默认关闭，向后兼容）
stream_to_sse_from_buffer — 自动读取 Last-Event-ID 请求头并写入 SSE id: 行；可选 replay_mode（none | resume | history）、late_observer（allow | deny）、terminal_shortcut（off | done-shortcut），默认与现网一致
ReplayMode / LateObserver / TerminalShortcut / resolve_buffer_subscribe_plan — D31 buffer 回放策略（buffer_replay_policy）
SseTransportProfile / VALID_PROFILES / resolve_buffer_sse_options — D32 四 preset → buffer SSE 参数 SSOT（replay_mode / late_observer / terminal_shortcut）；profile=None 默认 workflow-tail（D36 P5；replayable-workflow deprecated 过渡）
LastEventIdValidator / TaggedAgentEvent — 续传校验与带 id 事件载荷
EventBuffer / ingest_stream_to_buffer — 长任务「POST 触发 ingest + GET 订阅」
stream_to_sse(..., middlewares=[]) — 序列化前可选事件管道（默认空链与 MVP 一致）
permission_mode="raise" + create_permission_raise_setup / PermissionWaitRegistry — SDK 权限挂起 → permission-request 事件（需 agentx ≥ 0.7.9）
EventMiddleware / EventMiddlewarePipeline / apply_middlewares — 脱敏、trace 注入、debug 过滤
wrap_langgraph(graph, input_data, ...) → LangGraphEventStream（内置 DEFAULT_UI_ADAPTER_CONFIG）
wrap_agent_session(session_stream, *, cancel=None) → AgentSessionEventStream；cancel 为可选 SessionCancelHandle
safe_aclose_events / EventSourceLifecycle — 上游释放（断连 finally 使用）
DEFAULT_UI_ADAPTER_CONFIG / merge_ui_adapter_config() — UI 适配器默认（UiAdapterConfigMerger）
PROTOCOL_VERSION — 当前为 "1"

AgentSession 取消语义

断连或 stream_to_sse 结束时，薄层在 finally 调用 events.aclose()。对 AgentSession 路径：

若传入 cancel=SessionCancelHandle，aclose() 优先 await cancel.cancel()
否则回退 await session_stream.aclose()

推荐在业务侧传入显式 cancel，以便 SDK 升级时语义稳定：

events = wrap_agent_session(session.astream(question), cancel=session)
return stream_to_sse(events, request=request)

EventMiddleware 示例（tool-result 脱敏）

import dataclasses
from langchain_agentx.observability.events.langchain_agentx_event_adapter import (
    LangchainAgentEvent,
    LangchainAgentEventType,
)

def redact_tool_result(event: LangchainAgentEvent) -> LangchainAgentEvent | None:
    if event.event_type != LangchainAgentEventType.TOOL_RESULT:
        return event
    data = dict(event.data)
    data["content"] = "[REDACTED]"
    if isinstance(data.get("display"), dict):
        display = dict(data["display"])
        display["value"] = "[REDACTED]"
        data["display"] = display
    return dataclasses.replace(event, data=data)

return stream_to_sse(events, request=request, middlewares=[redact_tool_result])

middleware 返回 None 表示丢弃该事件；抛错则转为 thin_layer error 终态帧。

permission_mode="raise"（V2-D，特性开关）

默认 permission_mode="bypass" 与 MVP 一致。编码类场景可启用 raise，将 SDK PermissionPromptHandler 挂起转为 documented extension 事件 permission-request，等待业务回写后继续。

依赖：langchain-agentx >= 0.7.9（L3 PermissionPromptHandler）。

from fastapi import FastAPI, HTTPException, Request
from langchain_agentx.loop.run_context import create_task_with_copied_context
from langchain_agentx_stream_ui_backend import (
    EventBuffer,
    PermissionWaitRegistry,
    create_permission_raise_setup,
    ingest_stream_to_buffer,
    stream_to_sse_from_buffer,
    wrap_langgraph,
)

app = FastAPI()
_BUFFERS: dict[str, EventBuffer] = {}
_PERMISSIONS: dict[str, PermissionWaitRegistry] = {}

@app.post("/wiki/{task_id}/start")
async def start(task_id: str):
    perm = create_permission_raise_setup(session_id=task_id)
    _PERMISSIONS[task_id] = perm.registry
    buf = EventBuffer()
    _BUFFERS[task_id] = buf
    graph = build_graph(services={"prompt_handler": perm.handler})  # 注入 agentx graph
    events = wrap_langgraph(graph, {"task_id": task_id}, permission_broker=perm.broker)
    # SDK >= 1.9.4（含 1.9.3 run_ctx + 1.9.4 closeout 诊断）：后台 ingest 须 copy_context
    create_task_with_copied_context(
        ingest_stream_to_buffer(perm.wrap_stream(events), buf),
        name=f"wiki-ingest-{task_id}",
    )
    return {"task_id": task_id}

@app.post("/wiki/{task_id}/permission")
async def permission(task_id: str, body: dict):
    registry = _PERMISSIONS[task_id]
    ok = await registry.resolve(body["request_id"], body)
    if not ok:
        raise HTTPException(404)
    return {"ok": True}

@app.get("/wiki/{task_id}/stream")
async def stream(task_id: str, request: Request):
    return stream_to_sse_from_buffer(
        _BUFFERS[task_id],
        request=request,
        permission_mode="raise",
        session_id=task_id,
    )

回退至 bypass：不传 prompt_handler，stream_to_sse(..., permission_mode="bypass")（默认）。

契约 Fixture

SSOT 路径：backend/tests/fixtures/sse/*.events.json。生成说明见该目录 README.md。

Metadata

Release files for langchain-agentx-stream-ui-backend 0.8.8

For a detailed explanation of source distributions (sdists) and built distributions (wheels), please see the package formats documentation.

Source distribution (sdist)

Source distribution for langchain-agentx-stream-ui-backend 0.8.8
File	Size	Uploaded	Provenance
langchain_agentx_stream_ui_backend-0.8.8.tar.gz	94.4 kB	Oct 21, 2026	Without provenance	Details

Built distribution (wheel)

Table of built distributions (wheels) for langchain-agentx-stream-ui-backend 0.8.8
File	Interpreter	ABI	Platform	Reset
langchain_agentx_stream_ui_backend-0.8.8-py3-none-any.whl 124.5 kB Oct 21, 2026 Without provenance	Python 3	none	any	Details

Total release size: 219.0 kB

Release files / langchain_agentx_stream_ui_backend-0.8.8.tar.gz

Download URL	langchain_agentx_stream_ui_backend-0.8.8.tar.gz
Size	94.4 kB
Tags	Source
SHA-256 checksum How to use checksums	`b1a7937cbf1138da771959ef8719eb7c5c61662cfae8a4db0afff118d68fd82d`
BLAKE2b-256 checksum How to use checksums	`b8b35283559558e048dd8ff4cd29006c5bc084a9e155d3ba685734d5f6e92d8b`
Upload date	Oct 21, 2026
Uploaded using Trusted Publishing? What is trusted publishing?	No
Uploaded via	`twine/6.2.0 CPython/3.11.9`

Release files / langchain_agentx_stream_ui_backend-0.8.8-py3-none-any.whl

Download URL	langchain_agentx_stream_ui_backend-0.8.8-py3-none-any.whl
Size	124.5 kB
Tags	Python 3
SHA-256 checksum How to use checksums	`85f5219a45d39ecf2f3c982a19ef8c648818abbe6d564d8d3b227cdb80455393`
BLAKE2b-256 checksum How to use checksums	`3034786dbf50509a2b8978d43cb17ab73a03a840bd01fa1c89a3ad21737fa9f3`
Upload date	Oct 21, 2026
Uploaded using Trusted Publishing? What is trusted publishing?	No
Uploaded via	`twine/6.2.0 CPython/3.11.9`

Security for langchain-agentx-stream-ui-backend 0.8.8

Report project as malware

Provenance

Provenance describes where a file came from. On PyPI, provenance is shared via attestations, which provide a verifiable record of the build or publishing details. View details, limitations and caveats.

No provenance found

None of the files in this release were uploaded with provenance metadata.

A note on package security

PyPI is a repository for distributing packages, not a service for auditing them. While third-party security researchers and our staff do their best to detect and remove malicious packages, PyPI cannot guarantee the quality, security or suitability of any given package.

To learn more about protecting yourself from malicious actors, visit our package security documentation.