콘텐츠로 이동

spakky-agui

spakky-agui는 spakky-agent의 protocol-neutral AgentEvent stream을 AG-UI 이벤트로 투영하고 FastAPI SSE, HTTP streaming, WebSocket, stdio 경계로 노출합니다.

FastAPI SSE/HTTP streaming/WebSocket endpoint는 @AGUICompatible @Agent 선언을 AgUiAgentRegistry에 등록한 뒤 post-processor가 host FastAPI Pod에 자동 mount합니다. add_agui_endpoint 계열 helper는 lower-level 호환 API입니다.

Public API

AG-UI protocol adapter plugin for Spakky Agent.

AgUiAgent = AGUICompatible module-attribute

Deprecated alias for :class:AGUICompatible.

PLUGIN_NAME = Plugin(name='spakky-agui') module-attribute

Plugin identifier for the AG-UI adapter package.

RunDriverFactory = Callable[[RunAgentInput, AgUiRunAgentInput, str | None], AsyncIterable[str]]

Resolves the agent run for a request and returns a ready SSE driver.

Receives the mapped core RunAgentInput, the raw AG-UI input (so the factory can ingest an approval decision against its own signal repository), and the request Accept header for the event encoder.

AgUiConfig()

Bases: BaseSettings

Settings for the AG-UI adapter.

Source code in plugins/spakky-agui/src/spakky/plugins/agui/config.py 
50
51
def __init__(self) -> None:
    super().__init__()

sse_path = DEFAULT_AGUI_SSE_PATH class-attribute instance-attribute

Path the AG-UI SSE endpoint is mounted at on the FastAPI application.

websocket_path = DEFAULT_AGUI_WEBSOCKET_PATH class-attribute instance-attribute

Path the AG-UI WebSocket endpoint is mounted at on the FastAPI application.

http_stream_path = DEFAULT_AGUI_HTTP_STREAM_PATH class-attribute instance-attribute

Path the AG-UI HTTP streaming endpoint is mounted at on the FastAPI app.

emit_state_snapshot = True class-attribute instance-attribute

Whether STATE_SNAPSHOT neutral events are projected to AG-UI; when False the projector drops them so a client that ignores shared state is not sent redundant snapshots.

messages_snapshot_enabled = False class-attribute instance-attribute

Whether a single MESSAGES_SNAPSHOT is emitted before RUN_FINISHED; the framework runner emits no message history, so this defaults off and is only enabled by a client that wants a (currently empty) snapshot frame.

AbstractAgUiError

Bases: AbstractSpakkyFrameworkError, ABC

Base class for AG-UI adapter errors.

AgUiApprovalDecodeError

Bases: AbstractAgUiError

Raised when a resume input claims an approval decision it cannot supply.

The AG-UI resume carries an approval decision either as a tool-result message addressed to the deferred hitl_approval call or as a forwardedProps.approvalDecision object. This error is raised when that payload is present but malformed: the request id is missing, the decision string is absent, or the decision is not a member of ApprovalDecision.

AgUiEndpointConflictError

Bases: AbstractAgUiError

Raised when multiple AG-UI agents claim the same transport path.

AgUiPendingApprovalError

Bases: AbstractAgUiError

Raised when a paused-for-approval state carries malformed approval metadata.

The event-driven path converts RunPausedEvent to a deferred-tool approval request. Legacy helpers can still rebuild that request from durable WAIT_FOR_APPROVAL state. This error is raised when either source lacks the approval id, prompt, or known decision list the adapter needs to render the pause without guessing.

AgUiRunResolutionError

Bases: AbstractAgUiError

Raised when an SSE request references a run the driver cannot resolve.

The endpoint maps an AG-UI RunAgentInput to a core run, then asks the run-driver factory to build a driver for it. This error is raised when the factory cannot produce a runner for the requested agent/run — for example, the AG-UI input omits the last user message the core run requires to seed a model request.

AgUiProjector(config)

Stateful per-run projector from neutral events to AG-UI events.

Source code in plugins/spakky-agui/src/spakky/plugins/agui/projector.py 
110
111
112
113
114
115
116
def __init__(self, config: AgUiConfig) -> None:
    self._config = config
    self._open_message_id: str | None = None
    self._open_reasoning_id: str | None = None
    # Insertion-ordered set of in-flight tool-call ids (dict keys, value unused)
    # so a truncated stream flushes END frames in a deterministic open order.
    self._open_tool_call_ids: dict[str, None] = {}

project(event)

Project one neutral event into zero or more AG-UI events.

The neutral kind field is typed AgentEventKind (not a per-class Literal), so it does not narrow the union; matching on the event type does, which keeps each handler statically typed without casts.

Source code in plugins/spakky-agui/src/spakky/plugins/agui/projector.py 
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
def project(self, event: AgentEvent) -> list[BaseEvent]:
    """Project one neutral event into zero or more AG-UI events.

    The neutral ``kind`` field is typed ``AgentEventKind`` (not a per-class
    ``Literal``), so it does not narrow the union; matching on the event
    *type* does, which keeps each handler statically typed without casts.
    """
    match event:
        case NeutralMessageDeltaEvent():
            return self._project_message_delta(event)
        case NeutralReasoningDeltaEvent():
            return self._project_reasoning_delta(event)
        case NeutralToolCallStartEvent():
            return self._project_tool_start(event)
        case NeutralToolCallArgsDeltaEvent():
            return self._project_tool_args(event)
        case NeutralToolCallEndEvent():
            return self._project_tool_end(event)
        case NeutralToolCallResultEvent():
            return self._project_tool_result(event)
        case NeutralRunStartedEvent():
            return self._project_run_started(event)
        case NeutralRunPausedEvent():
            return self._project_run_paused(event)
        case NeutralRunFinishedEvent():
            return self._project_run_finished(event)
        case NeutralStepStartedEvent():
            return self._project_step_started(event)
        case NeutralStepFinishedEvent():
            return self._project_step_finished(event)
        case NeutralStateSnapshotEvent():
            return self._project_state_snapshot(event)
        case NeutralStateDeltaEvent():
            return self._project_state_delta(event)
        case NeutralArtifactEvent():  # pragma: no branch - exhaustive AgentEvent union
            return self._project_artifact(event)

finish()

Flush any open message, reasoning, or tool frames as END events.

Called once after the neutral stream ends so a stream truncated mid-frame (no RUN_FINISHED, or a model that stopped mid-message) still produces a balanced AG-UI sequence on the wire.

Source code in plugins/spakky-agui/src/spakky/plugins/agui/projector.py 
155
156
157
158
159
160
161
162
def finish(self) -> list[BaseEvent]:
    """Flush any open message, reasoning, or tool frames as END events.

    Called once after the neutral stream ends so a stream truncated mid-frame
    (no RUN_FINISHED, or a model that stopped mid-message) still produces a
    balanced AG-UI sequence on the wire.
    """
    return self._close_open_frames()

AgUiAgentEntry(instance, agent_type, metadata) dataclass

A discovered @Agent instance paired with its AG-UI metadata.

agent_name property

Return the stable AG-UI agent id.

AgUiAgentRegistry()

Holds AG-UI-exposed @Agent instances keyed by agent name.

Source code in plugins/spakky-agui/src/spakky/plugins/agui/server/registry.py 
33
34
def __init__(self) -> None:
    self._entries = {}

register(instance, agent_type, metadata)

Register one exposed @Agent instance and return the entry.

Source code in plugins/spakky-agui/src/spakky/plugins/agui/server/registry.py 
36
37
38
39
40
41
42
43
44
45
46
47
48
49
def register(
    self,
    instance: object,
    agent_type: type[object],
    metadata: AGUICompatible,
) -> AgUiAgentEntry:
    """Register one exposed @Agent instance and return the entry."""
    entry = AgUiAgentEntry(
        instance=instance,
        agent_type=agent_type,
        metadata=metadata,
    )
    self._entries[entry.agent_name] = entry
    return entry

get(agent_name)

Return the entry for agent_name.

Source code in plugins/spakky-agui/src/spakky/plugins/agui/server/registry.py 
51
52
53
54
55
56
def get(self, agent_name: str) -> AgUiAgentEntry:
    """Return the entry for ``agent_name``."""
    entry = self._entries.get(agent_name)
    if entry is None:
        raise AgUiRunResolutionError
    return entry

list_entries()

Return registered AG-UI agent entries in stable name order.

Source code in plugins/spakky-agui/src/spakky/plugins/agui/server/registry.py 
58
59
60
def list_entries(self) -> tuple[AgUiAgentEntry, ...]:
    """Return registered AG-UI agent entries in stable name order."""
    return tuple(self._entries[name] for name in sorted(self._entries))

AgUiStdioCommand(run_driver_factory, input_stream, output_stream, accept=None) dataclass

Callable command object that CLI plugins can register with their runner.

__call__(run_input_json=None) async

Run one AG-UI input from run_input_json or stdin over stdio.

Source code in plugins/spakky-agui/src/spakky/plugins/agui/stdio.py 
29
30
31
32
33
34
35
36
37
async def __call__(self, run_input_json: str | None = None) -> None:
    """Run one AG-UI input from ``run_input_json`` or stdin over stdio."""
    await run_agui_stdio(
        run_driver_factory=self.run_driver_factory,
        input_stream=self.input_stream,
        output_stream=self.output_stream,
        run_input_json=run_input_json,
        accept=self.accept,
    )

AGUICompatible(sse_path=None, http_stream_path=None, websocket_path=None) dataclass

Bases: Tag

Marks an @Agent class to be served through the AG-UI adapter.

Paths default to :class:AgUiConfig so a single declaration can use the plugin defaults, while multi-agent applications can assign distinct paths declaratively on each marked agent.

sse_path = None class-attribute instance-attribute

SSE endpoint path for this agent. None uses AgUiConfig.sse_path.

http_stream_path = None class-attribute instance-attribute

HTTP streaming endpoint path. None uses AgUiConfig.http_stream_path.

websocket_path = None class-attribute instance-attribute

WebSocket endpoint path. None uses AgUiConfig.websocket_path.

AgUiManagedRunDriver(runner_context, inbound, agent_id, config, accept)

Open a request-scoped runner for the lifetime of one AG-UI stream.

Source code in plugins/spakky-agui/src/spakky/plugins/agui/transport.py 
76
77
78
79
80
81
82
83
84
85
86
87
88
def __init__(
    self,
    runner_context: AbstractAsyncContextManager[AgentRunner],
    inbound: AgUiInboundRun,
    agent_id: str,
    config: AgUiConfig,
    accept: str | None,
) -> None:
    self._runner_context = runner_context
    self._inbound = inbound
    self._agent_id = agent_id
    self._config = config
    self._accept = accept

__aiter__() async

Yield frames while the runner factory context remains open.

Source code in plugins/spakky-agui/src/spakky/plugins/agui/transport.py 
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
async def __aiter__(self) -> AsyncIterator[str]:
    """Yield frames while the runner factory context remains open."""
    async with self._runner_context as runner:
        if self._inbound.core_input.resume:
            if runner.signals is None:
                raise AgUiApprovalDecodeError
            ingest_decision(
                self._inbound.ag_ui_input,
                runner.signals,
                self._inbound.core_input.state_id,
            )
        driver = AgUiRunDriver(
            runner=runner,
            run_input=self._inbound.core_input,
            agent_id=self._agent_id,
            projector=AgUiProjector(self._config),
            encoder=EventEncoder(accept=self._accept or ""),
        )
        async for frame in driver:
            yield frame

AgUiRunDriver(runner, run_input, agent_id, projector, encoder)

Streams one agent run as encoded AG-UI SSE frames.

Source code in plugins/spakky-agui/src/spakky/plugins/agui/transport.py 
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
def __init__(
    self,
    runner: AgentRunner,
    run_input: RunAgentInput,
    agent_id: str,
    projector: AgUiProjector,
    encoder: EventEncoder,
) -> None:
    self._runner = runner
    self._run_input = run_input
    self._attribution = AgentEventAttribution(
        agent_id=agent_id,
        run_id=run_input.state_id,
        conversation_id=run_input.effective_conversation_id,
    )
    self._projector = projector
    self._encoder = encoder

__aiter__() async

Yield SSE frames for the full run, including the flush tail.

Source code in plugins/spakky-agui/src/spakky/plugins/agui/transport.py 
56
57
58
59
60
61
62
63
64
async def __aiter__(self) -> AsyncIterator[str]:
    """Yield SSE frames for the full run, including the flush tail."""
    # Phase 1: project every neutral runner event.
    async for event in self._runner.run_events(self._run_input):
        for frame in self._frames_for(event):
            yield frame
    # Phase 2: flush any projector frame left open by a truncated stream.
    for frame in self._encode(self._projector.finish()):
        yield frame

add_agui_endpoint(app, *, run_driver_factory, config)

Register the AG-UI SSE endpoint on app at config.sse_path.

Source code in plugins/spakky-agui/src/spakky/plugins/agui/endpoint.py 
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
def add_agui_endpoint(
    app: FastAPI,
    *,
    run_driver_factory: RunDriverFactory,
    config: AgUiConfig,
) -> None:
    """Register the AG-UI SSE endpoint on ``app`` at ``config.sse_path``."""

    async def run_agui(request: Request) -> StreamingResponse:
        ag_ui_input = AgUiRunAgentInput.model_validate(await request.json())
        core_input = _to_core_input(ag_ui_input)
        driver = run_driver_factory(
            core_input,
            ag_ui_input,
            request.headers.get("accept"),
        )
        return StreamingResponse(driver, media_type=SSE_MEDIA_TYPE)

    app.add_api_route(config.sse_path, run_agui, methods=["POST"])

approval_from_pause(event)

Convert a neutral pause event into the AG-UI approval payload.

Source code in plugins/spakky-agui/src/spakky/plugins/agui/hitl.py 
88
89
90
91
92
93
94
95
96
97
98
99
def approval_from_pause(event: RunPausedEvent) -> Approval:
    """Convert a neutral pause event into the AG-UI approval payload."""
    if event.approval_id is None:
        raise AgUiPendingApprovalError
    return Approval(
        id=event.approval_id,
        prompt=event.prompt,
        allowed_decisions=tuple(
            _decode_decision(decision) for decision in event.allowed_decisions
        ),
        metadata=dict(event.metadata),
    )

ingest_decision(ag_ui_input, signals, state_id)

Decode an approval decision from an AG-UI input and queue it as a signal.

Reads the decision from the hitl_approval tool-result message when present, otherwise from forwardedProps.approvalDecision. The decoded decision is appended as an APPROVAL_DECISION signal carrying the request id the runner correlates against, plus optional modified payload and comment.

Source code in plugins/spakky-agui/src/spakky/plugins/agui/hitl.py 
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
def ingest_decision(
    ag_ui_input: AgUiRunAgentInput,
    signals: IAgentSignalRepository,
    state_id: str,
) -> None:
    """Decode an approval decision from an AG-UI input and queue it as a signal.

    Reads the decision from the ``hitl_approval`` tool-result message when
    present, otherwise from ``forwardedProps.approvalDecision``. The decoded
    decision is appended as an ``APPROVAL_DECISION`` signal carrying the request
    id the runner correlates against, plus optional modified payload and comment.
    """
    decision_payload = _extract_decision_payload(ag_ui_input)
    request_id = decision_payload.get("request_id")
    decision_value = decision_payload.get("decision")
    if not isinstance(request_id, str) or not isinstance(decision_value, str):
        raise AgUiApprovalDecodeError
    decision = _parse_decision(decision_value)
    payload: dict[str, JsonValue] = {
        "request_id": request_id,
        "decision": decision.value,
    }
    modified_payload = decision_payload.get("modified_payload")
    if modified_payload is not None:
        payload["modified_payload"] = modified_payload
    comment = decision_payload.get("comment")
    if comment is not None:
        payload["comment"] = comment
    signals.append(
        AgentSignal(
            id=f"agui-approval:{uuid.uuid4().hex}",
            agent_state_id=state_id,
            kind=AgentSignalKind.APPROVAL_DECISION,
            payload=payload,
        )
    )

project_approval(approval, attribution)

Render an approval request as a deferred-tool frame (no result).

The deferred call id is the approval id, so the resume tool-result message addresses the same call. The args carry the human-facing prompt, the allowed decisions, and any approval metadata the runner attached.

Source code in plugins/spakky-agui/src/spakky/plugins/agui/hitl.py 
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
def project_approval(
    approval: Approval,
    attribution: AgentEventAttribution,
) -> list[AgentEvent]:
    """Render an approval request as a deferred-tool frame (no result).

    The deferred call id is the approval id, so the resume tool-result message
    addresses the same call. The args carry the human-facing prompt, the allowed
    decisions, and any approval metadata the runner attached.
    """
    args: JsonObject = {
        "prompt": approval.prompt,
        "allowed_decisions": [
            decision.value for decision in approval.allowed_decisions
        ],
        **approval.metadata,
    }
    return [
        ToolCallStartEvent(
            attribution=attribution,
            call_id=approval.id,
            tool_name=HITL_APPROVAL_TOOL_NAME,
        ),
        ToolCallArgsDeltaEvent(
            attribution=attribution,
            call_id=approval.id,
            args_delta=dump_json(args),
        ),
        ToolCallEndEvent(attribution=attribution, call_id=approval.id),
    ]

project_pending_approval(state, attribution)

Project a durable pending approval into the deferred-tool request frame.

This remains as a compatibility helper for callers that already hold a durable state snapshot. The run driver consumes RunPausedEvent directly. Returns an empty list when the state is not paused for approval.

Source code in plugins/spakky-agui/src/spakky/plugins/agui/hitl.py 
127
128
129
130
131
132
133
134
135
136
137
138
139
140
def project_pending_approval(
    state: AgentState,
    attribution: AgentEventAttribution,
) -> list[AgentEvent]:
    """Project a durable pending approval into the deferred-tool request frame.

    This remains as a compatibility helper for callers that already hold a
    durable state snapshot. The run driver consumes ``RunPausedEvent`` directly.
    Returns an empty list when the state is not paused for approval.
    """
    approval = find_pending_approval(state)
    if approval is None:
        return []
    return project_approval(approval, attribution)

add_agui_http_stream_endpoint(app, *, run_driver_factory, config)

Register the AG-UI HTTP streaming endpoint at config.http_stream_path.

Source code in plugins/spakky-agui/src/spakky/plugins/agui/http_stream.py 
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
def add_agui_http_stream_endpoint(
    app: FastAPI,
    *,
    run_driver_factory: RunDriverFactory,
    config: AgUiConfig,
) -> None:
    """Register the AG-UI HTTP streaming endpoint at ``config.http_stream_path``."""

    async def run_agui_http_stream(request: Request) -> StreamingResponse:
        ag_ui_input = AgUiRunAgentInput.model_validate(await request.json())
        core_input = _to_core_input(ag_ui_input)
        driver = run_driver_factory(
            core_input,
            ag_ui_input,
            request.headers.get("accept"),
        )
        return StreamingResponse(
            _http_stream_chunks(driver),
            media_type=HTTP_STREAM_MEDIA_TYPE,
        )

    app.add_api_route(config.http_stream_path, run_agui_http_stream, methods=["POST"])

agui_stdio_payloads(driver) async

Yield AG-UI event JSON-lines from an AgUiRunDriver-compatible stream.

Source code in plugins/spakky-agui/src/spakky/plugins/agui/stdio.py 
70
71
72
73
async def agui_stdio_payloads(driver: AsyncIterable[str]) -> AsyncIterator[str]:
    """Yield AG-UI event JSON-lines from an ``AgUiRunDriver``-compatible stream."""
    async for frame in driver:
        yield sse_frame_payload(frame)

read_agui_run_input(*, input_stream, run_input_json=None)

Parse an AG-UI RunAgentInput from an argument or stdin.

Source code in plugins/spakky-agui/src/spakky/plugins/agui/stdio.py 
60
61
62
63
64
65
66
67
def read_agui_run_input(
    *,
    input_stream: TextIO,
    run_input_json: str | None = None,
) -> AgUiRunAgentInput:
    """Parse an AG-UI ``RunAgentInput`` from an argument or stdin."""
    payload = run_input_json if run_input_json is not None else input_stream.read()
    return AgUiRunAgentInput.model_validate_json(payload)

run_agui_stdio(*, run_driver_factory, input_stream, output_stream, run_input_json=None, accept=None) async

Drive one AG-UI run from stdio and write AG-UI event payloads to stdout.

Source code in plugins/spakky-agui/src/spakky/plugins/agui/stdio.py 
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
async def run_agui_stdio(
    *,
    run_driver_factory: RunDriverFactory,
    input_stream: TextIO,
    output_stream: TextIO,
    run_input_json: str | None = None,
    accept: str | None = None,
) -> None:
    """Drive one AG-UI run from stdio and write AG-UI event payloads to stdout."""
    ag_ui_input = read_agui_run_input(
        input_stream=input_stream,
        run_input_json=run_input_json,
    )
    core_input = _to_core_input(ag_ui_input)
    driver = run_driver_factory(core_input, ag_ui_input, accept)
    async for payload in agui_stdio_payloads(driver):
        output_stream.write(payload)
        output_stream.flush()

add_agui_websocket_endpoint(app, *, run_driver_factory, config)

Register the AG-UI WebSocket endpoint on app at config.websocket_path.

Source code in plugins/spakky-agui/src/spakky/plugins/agui/websocket.py 
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
def add_agui_websocket_endpoint(
    app: FastAPI,
    *,
    run_driver_factory: RunDriverFactory,
    config: AgUiConfig,
) -> None:
    """Register the AG-UI WebSocket endpoint on ``app`` at ``config.websocket_path``."""

    async def run_agui_websocket(websocket: WebSocket) -> None:
        await websocket.accept()
        try:
            while True:
                ag_ui_input = AgUiRunAgentInput.model_validate(
                    await websocket.receive_json()
                )
                core_input = _to_core_input(ag_ui_input)
                driver = run_driver_factory(
                    core_input,
                    ag_ui_input,
                    websocket.headers.get("accept"),
                )
                async for frame in driver:
                    await websocket.send_text(frame)
        except WebSocketDisconnect:
            return

    app.add_api_websocket_route(config.websocket_path, run_agui_websocket)

설정

Configuration for the spakky-agui plugin.

SPAKKY_AGUI_CONFIG_ENV_PREFIX = 'SPAKKY_AGUI_' module-attribute

Environment prefix for AG-UI adapter settings.

DEFAULT_AGUI_SSE_PATH = '/agui' module-attribute

Default mount path for the AG-UI SSE endpoint.

DEFAULT_AGUI_WEBSOCKET_PATH = '/agui/ws' module-attribute

Default mount path for the AG-UI WebSocket endpoint.

DEFAULT_AGUI_HTTP_STREAM_PATH = '/agui/stream' module-attribute

Default mount path for the AG-UI HTTP streaming endpoint.

AgUiConfig()

Bases: BaseSettings

Settings for the AG-UI adapter.

Source code in plugins/spakky-agui/src/spakky/plugins/agui/config.py 
50
51
def __init__(self) -> None:
    super().__init__()

sse_path = DEFAULT_AGUI_SSE_PATH class-attribute instance-attribute

Path the AG-UI SSE endpoint is mounted at on the FastAPI application.

websocket_path = DEFAULT_AGUI_WEBSOCKET_PATH class-attribute instance-attribute

Path the AG-UI WebSocket endpoint is mounted at on the FastAPI application.

http_stream_path = DEFAULT_AGUI_HTTP_STREAM_PATH class-attribute instance-attribute

Path the AG-UI HTTP streaming endpoint is mounted at on the FastAPI app.

emit_state_snapshot = True class-attribute instance-attribute

Whether STATE_SNAPSHOT neutral events are projected to AG-UI; when False the projector drops them so a client that ignores shared state is not sent redundant snapshots.

messages_snapshot_enabled = False class-attribute instance-attribute

Whether a single MESSAGES_SNAPSHOT is emitted before RUN_FINISHED; the framework runner emits no message history, so this defaults off and is only enabled by a client that wants a (currently empty) snapshot frame.

Endpoint

@AGUICompatible marker for exposing an @Agent over AG-UI transports.

AGUICompatible(sse_path=None, http_stream_path=None, websocket_path=None) dataclass

Bases: Tag

Marks an @Agent class to be served through the AG-UI adapter.

Paths default to :class:AgUiConfig so a single declaration can use the plugin defaults, while multi-agent applications can assign distinct paths declaratively on each marked agent.

sse_path = None class-attribute instance-attribute

SSE endpoint path for this agent. None uses AgUiConfig.sse_path.

http_stream_path = None class-attribute instance-attribute

HTTP streaming endpoint path. None uses AgUiConfig.http_stream_path.

websocket_path = None class-attribute instance-attribute

WebSocket endpoint path. None uses AgUiConfig.websocket_path.

Registry of @Agent instances exposed through AG-UI transports.

AgUiAgentEntry(instance, agent_type, metadata) dataclass

A discovered @Agent instance paired with its AG-UI metadata.

agent_name property

Return the stable AG-UI agent id.

AgUiAgentRegistry()

Holds AG-UI-exposed @Agent instances keyed by agent name.

Source code in plugins/spakky-agui/src/spakky/plugins/agui/server/registry.py 
33
34
def __init__(self) -> None:
    self._entries = {}

register(instance, agent_type, metadata)

Register one exposed @Agent instance and return the entry.

Source code in plugins/spakky-agui/src/spakky/plugins/agui/server/registry.py 
36
37
38
39
40
41
42
43
44
45
46
47
48
49
def register(
    self,
    instance: object,
    agent_type: type[object],
    metadata: AGUICompatible,
) -> AgUiAgentEntry:
    """Register one exposed @Agent instance and return the entry."""
    entry = AgUiAgentEntry(
        instance=instance,
        agent_type=agent_type,
        metadata=metadata,
    )
    self._entries[entry.agent_name] = entry
    return entry

get(agent_name)

Return the entry for agent_name.

Source code in plugins/spakky-agui/src/spakky/plugins/agui/server/registry.py 
51
52
53
54
55
56
def get(self, agent_name: str) -> AgUiAgentEntry:
    """Return the entry for ``agent_name``."""
    entry = self._entries.get(agent_name)
    if entry is None:
        raise AgUiRunResolutionError
    return entry

list_entries()

Return registered AG-UI agent entries in stable name order.

Source code in plugins/spakky-agui/src/spakky/plugins/agui/server/registry.py 
58
59
60
def list_entries(self) -> tuple[AgUiAgentEntry, ...]:
    """Return registered AG-UI agent entries in stable name order."""
    return tuple(self._entries[name] for name in sorted(self._entries))

Post-processor that mounts AG-UI routes on FastAPI Pods.

MountAgUiFastAPIPostProcessor()

Bases: IPostProcessor, IContainerAware, IApplicationContextAware

Discover @AGUICompatible @Agent Pods and mount their AG-UI FastAPI routes.

Source code in plugins/spakky-agui/src/spakky/plugins/agui/post_processors/mount_fastapi.py 
49
50
51
def __init__(self) -> None:
    self._claimed_paths = {}
    self._mounted = set()

post_process(pod)

Register exposed agents and mount them on FastAPI apps.

Source code in plugins/spakky-agui/src/spakky/plugins/agui/post_processors/mount_fastapi.py 
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
@override
def post_process(self, pod: object) -> object:
    """Register exposed agents and mount them on FastAPI apps."""
    if isinstance(pod, FastAPI):
        self._mount_registered_agents(pod)
        return pod
    agent_type = self._unwrap_proxy_type(type(pod))
    if not (AGUICompatible.exists(agent_type) and Agent.exists(agent_type)):
        return pod
    registry = self._container.get(AgUiAgentRegistry)
    entry = registry.register(pod, agent_type, AGUICompatible.get(agent_type))
    for app in self._fastapi_apps():
        self._mount_entry(app, entry)
    logger.info("Registered AG-UI agent from %s", agent_type.__qualname__)
    return pod

Mount the AG-UI SSE endpoint on a FastAPI application.

add_agui_endpoint registers a single POST {config.sse_path} route that accepts an AG-UI RunAgentInput and streams the run back as text/event-stream. It owns the protocol-boundary translation that neither the bridge nor the projector should know about: it maps the AG-UI input shape (threadId / runId / messages / parentRunId) onto the neutral core RunAgentInput and, when the input carries a resumed approval decision, queues that decision before the run is driven.

The application author supplies a run_driver_factory that resolves the concrete @Agent for the request and returns a ready AgUiRunDriver. The endpoint stays agnostic about which agent answers, mirroring pydantic-ai's add_*_fastapi_endpoint pattern and depending on third-party fastapi directly (ADR-0013 §2) rather than importing the spakky-fastapi plugin.

SSE_MEDIA_TYPE = 'text/event-stream' module-attribute

Media type for the AG-UI server-sent event stream.

RunDriverFactory = Callable[[RunAgentInput, AgUiRunAgentInput, str | None], AsyncIterable[str]]

Resolves the agent run for a request and returns a ready SSE driver.

Receives the mapped core RunAgentInput, the raw AG-UI input (so the factory can ingest an approval decision against its own signal repository), and the request Accept header for the event encoder.

add_agui_endpoint(app, *, run_driver_factory, config)

Register the AG-UI SSE endpoint on app at config.sse_path.

Source code in plugins/spakky-agui/src/spakky/plugins/agui/endpoint.py 
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
def add_agui_endpoint(
    app: FastAPI,
    *,
    run_driver_factory: RunDriverFactory,
    config: AgUiConfig,
) -> None:
    """Register the AG-UI SSE endpoint on ``app`` at ``config.sse_path``."""

    async def run_agui(request: Request) -> StreamingResponse:
        ag_ui_input = AgUiRunAgentInput.model_validate(await request.json())
        core_input = _to_core_input(ag_ui_input)
        driver = run_driver_factory(
            core_input,
            ag_ui_input,
            request.headers.get("accept"),
        )
        return StreamingResponse(driver, media_type=SSE_MEDIA_TYPE)

    app.add_api_route(config.sse_path, run_agui, methods=["POST"])

Mount the AG-UI HTTP chunked streaming endpoint on a FastAPI application.

add_agui_http_stream_endpoint registers a POST {config.http_stream_path} route that accepts an AG-UI RunAgentInput and streams encoded AG-UI event payloads as sequential JSON-line response chunks. It intentionally does not emit SSE data: framing; clients that want SSE should keep using add_agui_endpoint.

HTTP_STREAM_MEDIA_TYPE = 'application/x-ndjson' module-attribute

Media type for AG-UI HTTP streaming chunks.

add_agui_http_stream_endpoint(app, *, run_driver_factory, config)

Register the AG-UI HTTP streaming endpoint at config.http_stream_path.

Source code in plugins/spakky-agui/src/spakky/plugins/agui/http_stream.py 
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
def add_agui_http_stream_endpoint(
    app: FastAPI,
    *,
    run_driver_factory: RunDriverFactory,
    config: AgUiConfig,
) -> None:
    """Register the AG-UI HTTP streaming endpoint at ``config.http_stream_path``."""

    async def run_agui_http_stream(request: Request) -> StreamingResponse:
        ag_ui_input = AgUiRunAgentInput.model_validate(await request.json())
        core_input = _to_core_input(ag_ui_input)
        driver = run_driver_factory(
            core_input,
            ag_ui_input,
            request.headers.get("accept"),
        )
        return StreamingResponse(
            _http_stream_chunks(driver),
            media_type=HTTP_STREAM_MEDIA_TYPE,
        )

    app.add_api_route(config.http_stream_path, run_agui_http_stream, methods=["POST"])

Mount the AG-UI WebSocket endpoint on a FastAPI application.

add_agui_websocket_endpoint registers a bidirectional WebSocket route. Each client JSON message is parsed as an AG-UI RunAgentInput, mapped through the same protocol boundary as the SSE endpoint, and streamed back as encoded AG-UI event frames over WebSocket text messages.

add_agui_websocket_endpoint(app, *, run_driver_factory, config)

Register the AG-UI WebSocket endpoint on app at config.websocket_path.

Source code in plugins/spakky-agui/src/spakky/plugins/agui/websocket.py 
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
def add_agui_websocket_endpoint(
    app: FastAPI,
    *,
    run_driver_factory: RunDriverFactory,
    config: AgUiConfig,
) -> None:
    """Register the AG-UI WebSocket endpoint on ``app`` at ``config.websocket_path``."""

    async def run_agui_websocket(websocket: WebSocket) -> None:
        await websocket.accept()
        try:
            while True:
                ag_ui_input = AgUiRunAgentInput.model_validate(
                    await websocket.receive_json()
                )
                core_input = _to_core_input(ag_ui_input)
                driver = run_driver_factory(
                    core_input,
                    ag_ui_input,
                    websocket.headers.get("accept"),
                )
                async for frame in driver:
                    await websocket.send_text(frame)
        except WebSocketDisconnect:
            return

    app.add_api_websocket_route(config.websocket_path, run_agui_websocket)

AG-UI stdio protocol boundary for CLI adapters.

The stdio boundary intentionally emits protocol payloads only: it accepts a single AG-UI RunAgentInput JSON document from stdin or an argument, drives the same AgUiRunDriver used by SSE/HTTP/WebSocket adapters, and writes one AG-UI event JSON payload per stdout line. Rendering, colors, and nested views remain the responsibility of the consuming UI process.

AgUiStdioCommand(run_driver_factory, input_stream, output_stream, accept=None) dataclass

Callable command object that CLI plugins can register with their runner.

__call__(run_input_json=None) async

Run one AG-UI input from run_input_json or stdin over stdio.

Source code in plugins/spakky-agui/src/spakky/plugins/agui/stdio.py 
29
30
31
32
33
34
35
36
37
async def __call__(self, run_input_json: str | None = None) -> None:
    """Run one AG-UI input from ``run_input_json`` or stdin over stdio."""
    await run_agui_stdio(
        run_driver_factory=self.run_driver_factory,
        input_stream=self.input_stream,
        output_stream=self.output_stream,
        run_input_json=run_input_json,
        accept=self.accept,
    )

run_agui_stdio(*, run_driver_factory, input_stream, output_stream, run_input_json=None, accept=None) async

Drive one AG-UI run from stdio and write AG-UI event payloads to stdout.

Source code in plugins/spakky-agui/src/spakky/plugins/agui/stdio.py 
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
async def run_agui_stdio(
    *,
    run_driver_factory: RunDriverFactory,
    input_stream: TextIO,
    output_stream: TextIO,
    run_input_json: str | None = None,
    accept: str | None = None,
) -> None:
    """Drive one AG-UI run from stdio and write AG-UI event payloads to stdout."""
    ag_ui_input = read_agui_run_input(
        input_stream=input_stream,
        run_input_json=run_input_json,
    )
    core_input = _to_core_input(ag_ui_input)
    driver = run_driver_factory(core_input, ag_ui_input, accept)
    async for payload in agui_stdio_payloads(driver):
        output_stream.write(payload)
        output_stream.flush()

read_agui_run_input(*, input_stream, run_input_json=None)

Parse an AG-UI RunAgentInput from an argument or stdin.

Source code in plugins/spakky-agui/src/spakky/plugins/agui/stdio.py 
60
61
62
63
64
65
66
67
def read_agui_run_input(
    *,
    input_stream: TextIO,
    run_input_json: str | None = None,
) -> AgUiRunAgentInput:
    """Parse an AG-UI ``RunAgentInput`` from an argument or stdin."""
    payload = run_input_json if run_input_json is not None else input_stream.read()
    return AgUiRunAgentInput.model_validate_json(payload)

agui_stdio_payloads(driver) async

Yield AG-UI event JSON-lines from an AgUiRunDriver-compatible stream.

Source code in plugins/spakky-agui/src/spakky/plugins/agui/stdio.py 
70
71
72
73
async def agui_stdio_payloads(driver: AsyncIterable[str]) -> AsyncIterator[str]:
    """Yield AG-UI event JSON-lines from an ``AgUiRunDriver``-compatible stream."""
    async for frame in driver:
        yield sse_frame_payload(frame)

Transport

Drive an agent run and stream it as AG-UI server-sent events.

AgUiRunDriver is the pipe that connects the runner to the wire: it pulls the framework runner's native neutral AgentEvent stream off run_events — the lossless taxonomy AG-UI projects one-to-one (ADR-0013 §3) — runs each event through the projector (AgentEvent -> AG-UI BaseEvent), and encodes each AG-UI event into an SSE data: frame.

run_events emits approval pauses as neutral RunPausedEvent items. The projector maps those directly into AG-UI's deferred-tool request idiom. After the stream ends the driver flushes the projector's open-frame closures so a stream the runner left mid-message is still well-formed on the wire.

AgUiRunDriver(runner, run_input, agent_id, projector, encoder)

Streams one agent run as encoded AG-UI SSE frames.

Source code in plugins/spakky-agui/src/spakky/plugins/agui/transport.py 
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
def __init__(
    self,
    runner: AgentRunner,
    run_input: RunAgentInput,
    agent_id: str,
    projector: AgUiProjector,
    encoder: EventEncoder,
) -> None:
    self._runner = runner
    self._run_input = run_input
    self._attribution = AgentEventAttribution(
        agent_id=agent_id,
        run_id=run_input.state_id,
        conversation_id=run_input.effective_conversation_id,
    )
    self._projector = projector
    self._encoder = encoder

__aiter__() async

Yield SSE frames for the full run, including the flush tail.

Source code in plugins/spakky-agui/src/spakky/plugins/agui/transport.py 
56
57
58
59
60
61
62
63
64
async def __aiter__(self) -> AsyncIterator[str]:
    """Yield SSE frames for the full run, including the flush tail."""
    # Phase 1: project every neutral runner event.
    async for event in self._runner.run_events(self._run_input):
        for frame in self._frames_for(event):
            yield frame
    # Phase 2: flush any projector frame left open by a truncated stream.
    for frame in self._encode(self._projector.finish()):
        yield frame

AgUiManagedRunDriver(runner_context, inbound, agent_id, config, accept)

Open a request-scoped runner for the lifetime of one AG-UI stream.

Source code in plugins/spakky-agui/src/spakky/plugins/agui/transport.py 
76
77
78
79
80
81
82
83
84
85
86
87
88
def __init__(
    self,
    runner_context: AbstractAsyncContextManager[AgentRunner],
    inbound: AgUiInboundRun,
    agent_id: str,
    config: AgUiConfig,
    accept: str | None,
) -> None:
    self._runner_context = runner_context
    self._inbound = inbound
    self._agent_id = agent_id
    self._config = config
    self._accept = accept

__aiter__() async

Yield frames while the runner factory context remains open.

Source code in plugins/spakky-agui/src/spakky/plugins/agui/transport.py 
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
async def __aiter__(self) -> AsyncIterator[str]:
    """Yield frames while the runner factory context remains open."""
    async with self._runner_context as runner:
        if self._inbound.core_input.resume:
            if runner.signals is None:
                raise AgUiApprovalDecodeError
            ingest_decision(
                self._inbound.ag_ui_input,
                runner.signals,
                self._inbound.core_input.state_id,
            )
        driver = AgUiRunDriver(
            runner=runner,
            run_input=self._inbound.core_input,
            agent_id=self._agent_id,
            projector=AgUiProjector(self._config),
            encoder=EventEncoder(accept=self._accept or ""),
        )
        async for frame in driver:
            yield frame

Projection

Project neutral AgentEvents into AG-UI protocol events.

This is the fidelity-bearing half of the adapter. The neutral taxonomy carries deltas (MESSAGE_DELTA, REASONING_DELTA, TOOL_CALL_*), but AG-UI demands well-framed lifecycles: every text message is a TEXT_MESSAGE_START / …CONTENT / …END triple, every reasoning message a REASONING_START / REASONING_MESSAGE_START / …CONTENT / …END / REASONING_END sequence, every tool call a TOOL_CALL_START / …ARGS / …END (with the result as a separate TOOL_CALL_RESULT). The projector is the state machine that opens, continues, and closes those frames as deltas arrive.

It is stateful for the span of one run: it tracks the currently open message, the currently open reasoning message, and the set of open tool calls so that a delta with a new id closes the previous frame, and so finish() can flush any frame still open when the neutral stream ends (a truncated run stays well-formed on the wire).

ASSISTANT_ROLE = 'assistant' module-attribute

AG-UI text message role for model-authored assistant messages.

REASONING_ROLE = 'reasoning' module-attribute

AG-UI reasoning message role required by REASONING_MESSAGE_START.

ARTIFACT_CUSTOM_EVENT_NAME = 'artifact' module-attribute

CustomEvent name carrying a neutral artifact (no native AG-UI artifact).

AgUiProjector(config)

Stateful per-run projector from neutral events to AG-UI events.

Source code in plugins/spakky-agui/src/spakky/plugins/agui/projector.py 
110
111
112
113
114
115
116
def __init__(self, config: AgUiConfig) -> None:
    self._config = config
    self._open_message_id: str | None = None
    self._open_reasoning_id: str | None = None
    # Insertion-ordered set of in-flight tool-call ids (dict keys, value unused)
    # so a truncated stream flushes END frames in a deterministic open order.
    self._open_tool_call_ids: dict[str, None] = {}

project(event)

Project one neutral event into zero or more AG-UI events.

The neutral kind field is typed AgentEventKind (not a per-class Literal), so it does not narrow the union; matching on the event type does, which keeps each handler statically typed without casts.

Source code in plugins/spakky-agui/src/spakky/plugins/agui/projector.py 
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
def project(self, event: AgentEvent) -> list[BaseEvent]:
    """Project one neutral event into zero or more AG-UI events.

    The neutral ``kind`` field is typed ``AgentEventKind`` (not a per-class
    ``Literal``), so it does not narrow the union; matching on the event
    *type* does, which keeps each handler statically typed without casts.
    """
    match event:
        case NeutralMessageDeltaEvent():
            return self._project_message_delta(event)
        case NeutralReasoningDeltaEvent():
            return self._project_reasoning_delta(event)
        case NeutralToolCallStartEvent():
            return self._project_tool_start(event)
        case NeutralToolCallArgsDeltaEvent():
            return self._project_tool_args(event)
        case NeutralToolCallEndEvent():
            return self._project_tool_end(event)
        case NeutralToolCallResultEvent():
            return self._project_tool_result(event)
        case NeutralRunStartedEvent():
            return self._project_run_started(event)
        case NeutralRunPausedEvent():
            return self._project_run_paused(event)
        case NeutralRunFinishedEvent():
            return self._project_run_finished(event)
        case NeutralStepStartedEvent():
            return self._project_step_started(event)
        case NeutralStepFinishedEvent():
            return self._project_step_finished(event)
        case NeutralStateSnapshotEvent():
            return self._project_state_snapshot(event)
        case NeutralStateDeltaEvent():
            return self._project_state_delta(event)
        case NeutralArtifactEvent():  # pragma: no branch - exhaustive AgentEvent union
            return self._project_artifact(event)

finish()

Flush any open message, reasoning, or tool frames as END events.

Called once after the neutral stream ends so a stream truncated mid-frame (no RUN_FINISHED, or a model that stopped mid-message) still produces a balanced AG-UI sequence on the wire.

Source code in plugins/spakky-agui/src/spakky/plugins/agui/projector.py 
155
156
157
158
159
160
161
162
def finish(self) -> list[BaseEvent]:
    """Flush any open message, reasoning, or tool frames as END events.

    Called once after the neutral stream ends so a stream truncated mid-frame
    (no RUN_FINISHED, or a model that stopped mid-message) still produces a
    balanced AG-UI sequence on the wire.
    """
    return self._close_open_frames()

Human-in-the-loop projection and decision ingestion for the AG-UI adapter.

AG-UI has no first-class approval event, so an approval request surfaces as a deferred tool call: a TOOL_CALL_START/ARGS/END triple naming the synthetic hitl_approval tool, deliberately with no result frame. The client renders it, collects the human decision, and returns that decision on the next RunAgentInput (as the deferred tool's result message, or as forwardedProps.approvalDecision). ingest_decision decodes that decision and appends it to the durable signal queue the runner polls (ADR-0013 §5), which is the only channel the core run accepts an approval decision through.

The core run_events stream emits a first-class RunPausedEvent when a tool pauses for approval. AG-UI still represents that pause as a deferred tool call, but the adapter now projects the pause event directly instead of reconciling durable state after a terminal RUN_FINISHED.

HITL_APPROVAL_TOOL_NAME = 'hitl_approval' module-attribute

Synthetic tool name carrying an approval request as a deferred tool call.

APPROVAL_DECISION_FORWARDED_KEY = 'approvalDecision' module-attribute

forwardedProps key a client may use to return an approval decision.

APPROVAL_STATE_METADATA_KEY = 'approval' module-attribute

State-metadata key under which the runner stores the approval request.

project_approval(approval, attribution)

Render an approval request as a deferred-tool frame (no result).

The deferred call id is the approval id, so the resume tool-result message addresses the same call. The args carry the human-facing prompt, the allowed decisions, and any approval metadata the runner attached.

Source code in plugins/spakky-agui/src/spakky/plugins/agui/hitl.py 
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
def project_approval(
    approval: Approval,
    attribution: AgentEventAttribution,
) -> list[AgentEvent]:
    """Render an approval request as a deferred-tool frame (no result).

    The deferred call id is the approval id, so the resume tool-result message
    addresses the same call. The args carry the human-facing prompt, the allowed
    decisions, and any approval metadata the runner attached.
    """
    args: JsonObject = {
        "prompt": approval.prompt,
        "allowed_decisions": [
            decision.value for decision in approval.allowed_decisions
        ],
        **approval.metadata,
    }
    return [
        ToolCallStartEvent(
            attribution=attribution,
            call_id=approval.id,
            tool_name=HITL_APPROVAL_TOOL_NAME,
        ),
        ToolCallArgsDeltaEvent(
            attribution=attribution,
            call_id=approval.id,
            args_delta=dump_json(args),
        ),
        ToolCallEndEvent(attribution=attribution, call_id=approval.id),
    ]

approval_from_pause(event)

Convert a neutral pause event into the AG-UI approval payload.

Source code in plugins/spakky-agui/src/spakky/plugins/agui/hitl.py 
88
89
90
91
92
93
94
95
96
97
98
99
def approval_from_pause(event: RunPausedEvent) -> Approval:
    """Convert a neutral pause event into the AG-UI approval payload."""
    if event.approval_id is None:
        raise AgUiPendingApprovalError
    return Approval(
        id=event.approval_id,
        prompt=event.prompt,
        allowed_decisions=tuple(
            _decode_decision(decision) for decision in event.allowed_decisions
        ),
        metadata=dict(event.metadata),
    )

find_pending_approval(state)

Reconstruct the pending approval from a durable WAIT_FOR_APPROVAL state.

The runner saves the paused boundary as an INTERRUPTED state whose reason is APPROVAL_REQUIRED, carrying the AgentApprovalRequest metadata under metadata["approval"] and the human-facing prompt in current_activity. Returns the rebuilt approval when the state is paused for approval, otherwise None (the run finished or paused for some other reason).

Source code in plugins/spakky-agui/src/spakky/plugins/agui/hitl.py 
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
def find_pending_approval(state: AgentState) -> Approval | None:
    """Reconstruct the pending approval from a durable WAIT_FOR_APPROVAL state.

    The runner saves the paused boundary as an ``INTERRUPTED`` state whose reason
    is ``APPROVAL_REQUIRED``, carrying the ``AgentApprovalRequest`` metadata under
    ``metadata["approval"]`` and the human-facing prompt in ``current_activity``.
    Returns the rebuilt approval when the state is paused for approval, otherwise
    ``None`` (the run finished or paused for some other reason).
    """
    if (
        state.status is not AgentStatus.INTERRUPTED
        or state.reason is not AgentStateReason.APPROVAL_REQUIRED
    ):
        return None
    approval_metadata = state.metadata.get(APPROVAL_STATE_METADATA_KEY)
    if not isinstance(approval_metadata, Mapping) or state.current_activity is None:
        raise AgUiPendingApprovalError
    return Approval(
        id=_require_text(approval_metadata, "id"),
        prompt=state.current_activity,
        allowed_decisions=_decode_allowed_decisions(approval_metadata),
        metadata=_approval_request_metadata(approval_metadata),
    )

project_pending_approval(state, attribution)

Project a durable pending approval into the deferred-tool request frame.

This remains as a compatibility helper for callers that already hold a durable state snapshot. The run driver consumes RunPausedEvent directly. Returns an empty list when the state is not paused for approval.

Source code in plugins/spakky-agui/src/spakky/plugins/agui/hitl.py 
127
128
129
130
131
132
133
134
135
136
137
138
139
140
def project_pending_approval(
    state: AgentState,
    attribution: AgentEventAttribution,
) -> list[AgentEvent]:
    """Project a durable pending approval into the deferred-tool request frame.

    This remains as a compatibility helper for callers that already hold a
    durable state snapshot. The run driver consumes ``RunPausedEvent`` directly.
    Returns an empty list when the state is not paused for approval.
    """
    approval = find_pending_approval(state)
    if approval is None:
        return []
    return project_approval(approval, attribution)

ingest_decision(ag_ui_input, signals, state_id)

Decode an approval decision from an AG-UI input and queue it as a signal.

Reads the decision from the hitl_approval tool-result message when present, otherwise from forwardedProps.approvalDecision. The decoded decision is appended as an APPROVAL_DECISION signal carrying the request id the runner correlates against, plus optional modified payload and comment.

Source code in plugins/spakky-agui/src/spakky/plugins/agui/hitl.py 
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
def ingest_decision(
    ag_ui_input: AgUiRunAgentInput,
    signals: IAgentSignalRepository,
    state_id: str,
) -> None:
    """Decode an approval decision from an AG-UI input and queue it as a signal.

    Reads the decision from the ``hitl_approval`` tool-result message when
    present, otherwise from ``forwardedProps.approvalDecision``. The decoded
    decision is appended as an ``APPROVAL_DECISION`` signal carrying the request
    id the runner correlates against, plus optional modified payload and comment.
    """
    decision_payload = _extract_decision_payload(ag_ui_input)
    request_id = decision_payload.get("request_id")
    decision_value = decision_payload.get("decision")
    if not isinstance(request_id, str) or not isinstance(decision_value, str):
        raise AgUiApprovalDecodeError
    decision = _parse_decision(decision_value)
    payload: dict[str, JsonValue] = {
        "request_id": request_id,
        "decision": decision.value,
    }
    modified_payload = decision_payload.get("modified_payload")
    if modified_payload is not None:
        payload["modified_payload"] = modified_payload
    comment = decision_payload.get("comment")
    if comment is not None:
        payload["comment"] = comment
    signals.append(
        AgentSignal(
            id=f"agui-approval:{uuid.uuid4().hex}",
            agent_state_id=state_id,
            kind=AgentSignalKind.APPROVAL_DECISION,
            payload=payload,
        )
    )

carries_approval_decision(ag_ui_input)

Return whether the AG-UI input carries an approval decision to resume on.

Source code in plugins/spakky-agui/src/spakky/plugins/agui/hitl.py 
215
216
217
def carries_approval_decision(ag_ui_input: AgUiRunAgentInput) -> bool:
    """Return whether the AG-UI input carries an approval decision to resume on."""
    return _decision_source(ag_ui_input) is not None

Shared JSON serialization for AG-UI adapter frames.

The projector serializes tool results and run output, and HITL serializes approval args — both need the same neutral JsonValue -> JSON-text encoding, so it lives at module scope rather than being inlined twice.

dump_json(value)

Serialize a neutral JSON value to compact JSON text.

Source code in plugins/spakky-agui/src/spakky/plugins/agui/serialization.py 
13
14
15
def dump_json(value: JsonValue) -> str:
    """Serialize a neutral JSON value to compact JSON text."""
    return dumps(value, separators=(",", ":"), ensure_ascii=False)

sse_frame_payload(frame)

Convert one AG-UI SSE frame into a raw JSON-line event payload.

Source code in plugins/spakky-agui/src/spakky/plugins/agui/serialization.py 
18
19
20
21
22
23
24
25
def sse_frame_payload(frame: str) -> str:
    """Convert one AG-UI SSE frame into a raw JSON-line event payload."""
    payload_lines = [
        line.removeprefix("data:").lstrip()
        for line in frame.splitlines()
        if line.startswith("data:")
    ]
    return "\n".join(payload_lines) + "\n"

Plugin

Plugin initialization for the AG-UI adapter.

initialize(app)

Register AG-UI configuration, registry, and auto-mount post-processor.

Source code in plugins/spakky-agui/src/spakky/plugins/agui/main.py 
13
14
15
16
17
18
19
def initialize(app: SpakkyApplication) -> None:
    """Register AG-UI configuration, registry, and auto-mount post-processor."""
    if not app.container.contains(IAgentRunnerFactory):
        app.add(AgentRunnerFactory)
    app.add(AgUiConfig)
    app.add(AgUiAgentRegistry)
    app.add(MountAgUiFastAPIPostProcessor)

에러

Error classes for the spakky-agui plugin.

AbstractAgUiError

Bases: AbstractSpakkyFrameworkError, ABC

Base class for AG-UI adapter errors.

AgUiApprovalDecodeError

Bases: AbstractAgUiError

Raised when a resume input claims an approval decision it cannot supply.

The AG-UI resume carries an approval decision either as a tool-result message addressed to the deferred hitl_approval call or as a forwardedProps.approvalDecision object. This error is raised when that payload is present but malformed: the request id is missing, the decision string is absent, or the decision is not a member of ApprovalDecision.

AgUiPendingApprovalError

Bases: AbstractAgUiError

Raised when a paused-for-approval state carries malformed approval metadata.

The event-driven path converts RunPausedEvent to a deferred-tool approval request. Legacy helpers can still rebuild that request from durable WAIT_FOR_APPROVAL state. This error is raised when either source lacks the approval id, prompt, or known decision list the adapter needs to render the pause without guessing.

AgUiRunResolutionError

Bases: AbstractAgUiError

Raised when an SSE request references a run the driver cannot resolve.

The endpoint maps an AG-UI RunAgentInput to a core run, then asks the run-driver factory to build a driver for it. This error is raised when the factory cannot produce a runner for the requested agent/run — for example, the AG-UI input omits the last user message the core run requires to seed a model request.

AgUiEndpointConflictError

Bases: AbstractAgUiError

Raised when multiple AG-UI agents claim the same transport path.