wehub-resource-sync
978 lines
37 KiB
Python
978 lines
37 KiB
Python
"""Responses namespace — streaming, tool loop, polling, cancel, steer.
|
|
|
|
.. deprecated::
|
|
``ResponsesNamespace`` (``client.responses.*``) targets the
|
|
deprecated ``/v1/responses`` surface. New code should use
|
|
``SessionsNamespace`` (``client.sessions.*``) instead.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import inspect
|
|
import json
|
|
import logging
|
|
import warnings
|
|
from collections.abc import AsyncIterator
|
|
from typing import Any
|
|
|
|
import httpx
|
|
|
|
from ._errors import ToolCallDenied, raise_for_status, require_json_object, response_body
|
|
from ._events import (
|
|
ClientTaskCancel,
|
|
CompactionInProgress,
|
|
ElicitationRequest,
|
|
ErrorEvent,
|
|
MessageDone,
|
|
NativeToolCall,
|
|
OutputFileDone,
|
|
ReasoningDelta,
|
|
ReasoningStarted,
|
|
ReasoningSummaryDelta,
|
|
ResponseCancelled,
|
|
ResponseCompleted,
|
|
ResponseCreated,
|
|
ResponseFailed,
|
|
ResponseIncomplete,
|
|
RetryEvent,
|
|
StreamEvent,
|
|
TextDelta,
|
|
ToolCall,
|
|
ToolResult,
|
|
)
|
|
from ._sse import parse_sse_stream
|
|
from ._tool_handler import (
|
|
CompactionStartCtx,
|
|
ElicitationRequestCtx,
|
|
FileOutputCtx,
|
|
MessageEndCtx,
|
|
MessageStartCtx,
|
|
NativeToolCallCtx,
|
|
ReasoningEndCtx,
|
|
ResponseEndCtx,
|
|
ResponseStartCtx,
|
|
RetryCtx,
|
|
ServerErrorCtx,
|
|
StreamHooks,
|
|
ToolCallEndCtx,
|
|
ToolCallInfo,
|
|
ToolCallStartCtx,
|
|
ToolHandler,
|
|
ToolResultInfo,
|
|
ToolResultsReadyCtx,
|
|
)
|
|
from ._tool_handler import (
|
|
ReasoningStartCtx as ReasoningStartHookCtx,
|
|
)
|
|
from ._types import Response
|
|
|
|
_log = logging.getLogger("omnigent_client.responses")
|
|
|
|
# Terminal statuses — the response won't change further.
|
|
_TERMINAL_STATUSES = frozenset({"completed", "failed", "incomplete", "cancelled"})
|
|
|
|
# Async dispatch of client tools flows through the
|
|
# action_required handler (``_execute_and_patch``) — same path
|
|
# the sub-agent tunnel uses. The SDK tracks the spawned local
|
|
# tasks by call_id so :class:`ClientTaskCancel` events can find
|
|
# and cancel them; everything else is fire-and-forget and
|
|
# self-cleans on done.
|
|
|
|
|
|
async def _call_hook(hook: Any, ctx: Any) -> Any:
|
|
"""Call a hook (sync or async) and return its result."""
|
|
if hook is None:
|
|
return None
|
|
result = hook(ctx)
|
|
if inspect.isawaitable(result):
|
|
return await result
|
|
return result
|
|
|
|
|
|
class ResponsesNamespace:
|
|
"""Methods for ``/v1/responses`` endpoints.
|
|
|
|
.. deprecated::
|
|
This namespace targets the deprecated ``/v1/responses``
|
|
surface. Use :class:`SessionsNamespace`
|
|
(``client.sessions.*``) instead — see
|
|
``designs/SESSION_REARCHITECTURE.md``.
|
|
"""
|
|
|
|
def __init__(self, http: httpx.AsyncClient, base_url: str) -> None:
|
|
self._http = http
|
|
self._base = base_url
|
|
self._warned = False
|
|
|
|
def _deprecation_warn(self, method: str) -> None:
|
|
"""Emit a one-shot ``DeprecationWarning`` per instance.
|
|
|
|
:param method: Name of the deprecated method, e.g.
|
|
``"create"``.
|
|
"""
|
|
if not self._warned:
|
|
warnings.warn(
|
|
f"ResponsesNamespace.{method}() is deprecated; use client.sessions.* instead",
|
|
DeprecationWarning,
|
|
stacklevel=3,
|
|
)
|
|
self._warned = True
|
|
|
|
async def create(
|
|
self,
|
|
*,
|
|
model: str,
|
|
input: str | list[dict[str, object]],
|
|
background: bool = False,
|
|
instructions: str | None = None,
|
|
previous_response_id: str | None = None,
|
|
tools: list[dict[str, object]] | None = None,
|
|
reasoning: dict[str, str] | None = None,
|
|
model_override: str | None = None,
|
|
) -> Response:
|
|
"""Create a response (blocking, non-streaming).
|
|
|
|
:param model: Agent name.
|
|
:param input: User text or content block list.
|
|
:param background: If True, returns immediately (poll via get()).
|
|
:param instructions: Per-request system instructions.
|
|
:param previous_response_id: Prior response for multi-turn.
|
|
:param tools: Client-specified tool schemas.
|
|
:param reasoning: Reasoning config, e.g. ``{"effort": "high"}``.
|
|
:param model_override: Optional per-request LLM model override,
|
|
e.g. ``"openai/gpt-5.4-mini"``. Shadows the spec model.
|
|
:returns: The response object.
|
|
"""
|
|
self._deprecation_warn("create")
|
|
body = _build_body(
|
|
model=model,
|
|
input=input,
|
|
stream=False,
|
|
background=background,
|
|
instructions=instructions,
|
|
previous_response_id=previous_response_id,
|
|
tools=tools,
|
|
reasoning=reasoning,
|
|
model_override=model_override,
|
|
)
|
|
resp = await self._http.post(f"{self._base}/v1/responses", json=body)
|
|
raise_for_status(resp.status_code, response_body(resp))
|
|
return Response.from_dict(require_json_object(resp, "POST /v1/responses"))
|
|
|
|
async def stream(
|
|
self,
|
|
*,
|
|
model: str,
|
|
input: str | list[dict[str, object]],
|
|
background: bool = False,
|
|
instructions: str | None = None,
|
|
previous_response_id: str | None = None,
|
|
tool_handler: ToolHandler | None = None,
|
|
hooks: StreamHooks | None = None,
|
|
reasoning: dict[str, str] | None = None,
|
|
model_override: str | None = None,
|
|
) -> AsyncIterator[StreamEvent]:
|
|
"""Stream a response, optionally running the tool loop.
|
|
|
|
If ``tool_handler`` is provided, the client runs the full tool
|
|
execution loop: stream -> detect tool calls -> execute via handler
|
|
-> POST results -> stream again. The consumer sees one continuous
|
|
sequence of events.
|
|
|
|
If ``tool_handler`` is None, yields raw events for a single server
|
|
response.
|
|
|
|
:param model: Agent name.
|
|
:param input: User text or content block list.
|
|
:param tool_handler: Optional client-side tool execution config.
|
|
:param hooks: Optional lifecycle hooks.
|
|
"""
|
|
self._deprecation_warn("stream")
|
|
if hooks is None:
|
|
hooks = StreamHooks()
|
|
|
|
current_input: str | list[dict[str, object]] = input
|
|
current_prev_id = previous_response_id
|
|
iteration = 0
|
|
# Track local asyncio.Tasks spawned by the action_required
|
|
# handler so :class:`ClientTaskCancel` SSE events can find
|
|
# and cancel them. Keyed by ``call_id`` because that's the
|
|
# field the cancel event carries (looked up from the
|
|
# server-side pending_tool_calls row when the cancel fires).
|
|
# Persists across the outer ``while True`` so a task
|
|
# dispatched in one iteration can be cancelled by an event
|
|
# arriving in a later iteration's stream. Each entry
|
|
# self-cleans on done via ``add_done_callback``.
|
|
pending_local_tasks: dict[str, asyncio.Task[None]] = {}
|
|
|
|
while True:
|
|
tools = tool_handler.schemas if tool_handler is not None else None
|
|
pending_client_calls: list[ToolCall] = []
|
|
completed_call_ids: set[str] = set()
|
|
current_response_id: str | None = None
|
|
current_session_id: str | None = None
|
|
in_reasoning = False
|
|
reasoning_text = ""
|
|
summary_text = ""
|
|
message_started = False
|
|
|
|
body = _build_body(
|
|
model=model,
|
|
input=current_input,
|
|
stream=True,
|
|
background=background,
|
|
instructions=instructions,
|
|
previous_response_id=current_prev_id,
|
|
tools=tools,
|
|
reasoning=reasoning,
|
|
model_override=model_override,
|
|
)
|
|
|
|
async with self._http.stream("POST", f"{self._base}/v1/responses", json=body) as resp:
|
|
if resp.status_code >= 400:
|
|
await resp.aread()
|
|
raise_for_status(resp.status_code, response_body(resp))
|
|
|
|
async for event in parse_sse_stream(resp.aiter_bytes()):
|
|
# ── Fire hooks and collect state ──────────
|
|
|
|
if isinstance(event, ResponseCreated):
|
|
current_response_id = event.response.id
|
|
if event.response.conversation is not None:
|
|
current_session_id = event.response.conversation.id
|
|
await _call_hook(
|
|
hooks.on_response_start, ResponseStartCtx(response=event.response)
|
|
)
|
|
|
|
elif isinstance(event, ReasoningStarted):
|
|
in_reasoning = True
|
|
reasoning_text = ""
|
|
summary_text = ""
|
|
await _call_hook(hooks.on_reasoning_start, ReasoningStartHookCtx())
|
|
|
|
elif isinstance(event, ReasoningDelta):
|
|
reasoning_text += event.delta
|
|
|
|
elif isinstance(event, ReasoningSummaryDelta):
|
|
summary_text += event.delta
|
|
|
|
elif isinstance(event, TextDelta):
|
|
if in_reasoning:
|
|
in_reasoning = False
|
|
await _call_hook(
|
|
hooks.on_reasoning_end,
|
|
ReasoningEndCtx(
|
|
reasoning_text=reasoning_text, summary_text=summary_text
|
|
),
|
|
)
|
|
if not message_started:
|
|
message_started = True
|
|
await _call_hook(
|
|
hooks.on_message_start,
|
|
MessageStartCtx(response_id=current_response_id or ""),
|
|
)
|
|
|
|
elif isinstance(event, CompactionInProgress):
|
|
await _call_hook(hooks.on_compaction_start, CompactionStartCtx())
|
|
|
|
elif isinstance(event, ToolCall):
|
|
is_client_side = (
|
|
tool_handler is not None and event.call_id not in completed_call_ids
|
|
)
|
|
executed_by = "client" if is_client_side else "server"
|
|
await _call_hook(
|
|
hooks.on_tool_call_start,
|
|
ToolCallStartCtx(
|
|
name=event.name,
|
|
arguments=event.arguments,
|
|
call_id=event.call_id,
|
|
agent_name=event.agent_name,
|
|
executed_by=executed_by,
|
|
),
|
|
)
|
|
# Action_required tool calls (sub-agent
|
|
# tunnel + sys_call_async-dispatched
|
|
# client tools) execute locally and PATCH
|
|
# back while the stream continues.
|
|
# Tracked by call_id so a follow-up
|
|
# ``ClientTaskCancel`` event can cancel
|
|
# the local task.
|
|
if event.status == "action_required" and tool_handler is not None:
|
|
completed_call_ids.add(event.call_id)
|
|
local_task = asyncio.create_task(
|
|
_execute_and_patch(
|
|
self._http,
|
|
self._base,
|
|
tool_handler,
|
|
hooks,
|
|
event,
|
|
current_response_id or "",
|
|
iteration,
|
|
)
|
|
)
|
|
pending_local_tasks[event.call_id] = local_task
|
|
# Auto-clean once the task settles —
|
|
# PATCH back, exception, or cancel all
|
|
# land here. Captured in a closure so
|
|
# the dict reference stays live.
|
|
_local_task_call_id = event.call_id
|
|
|
|
def _drop_done(
|
|
_t: asyncio.Task[None],
|
|
_key: str = _local_task_call_id,
|
|
) -> None:
|
|
pending_local_tasks.pop(_key, None)
|
|
|
|
local_task.add_done_callback(_drop_done)
|
|
elif is_client_side:
|
|
# Synchronous client-side tool calls —
|
|
# the LLM called the tool directly.
|
|
# The SDK doesn't run them inline; it
|
|
# accumulates them and the outer
|
|
# ``stream()`` loop runs them after
|
|
# the response terminates and PATCHes
|
|
# ``tool_results`` to feed the next
|
|
# iteration. Authors who want
|
|
# background dispatch use
|
|
# ``sys_call_async`` instead, which
|
|
# produces an action_required event
|
|
# handled by the branch above.
|
|
pending_client_calls.append(event)
|
|
|
|
elif isinstance(event, ElicitationRequest):
|
|
# MCP-shape elicitation — the server is
|
|
# waiting on a verdict. Route to the
|
|
# elicitation hook (not the tool_handler —
|
|
# this is not a tool call). If no hook is
|
|
# registered, fail-closed decline so the
|
|
# parked workflow doesn't stall forever.
|
|
# Fire-and-forget elicitation response —
|
|
# see the peer tunneled-call site above for
|
|
# why we intentionally drop the task
|
|
# reference.
|
|
asyncio.ensure_future( # noqa: RUF006 — see comment above
|
|
_handle_elicitation_request(
|
|
self._http,
|
|
self._base,
|
|
hooks,
|
|
event,
|
|
current_response_id or "",
|
|
current_session_id or "",
|
|
)
|
|
)
|
|
|
|
elif isinstance(event, ToolResult):
|
|
completed_call_ids.add(event.call_id)
|
|
await _call_hook(
|
|
hooks.on_tool_call_end,
|
|
ToolCallEndCtx(
|
|
name="",
|
|
call_id=event.call_id,
|
|
agent_name="",
|
|
output=event.output,
|
|
),
|
|
)
|
|
|
|
elif isinstance(event, ClientTaskCancel):
|
|
# Server is telling us to stop the local
|
|
# body. Look up the local task by
|
|
# call_id (the server includes the
|
|
# synthetic call_id from the
|
|
# pending_tool_call row in the SSE
|
|
# payload) and cancel — the body's
|
|
# ``except CancelledError`` runs to
|
|
# cleanup. ``ensure_future``-style fires
|
|
# without a tracked task remain
|
|
# un-cancellable; that case is benign
|
|
# because the server has already decided
|
|
# the task is cancelled and the late
|
|
# PATCH back will be a G3 no-op.
|
|
if event.call_id is not None:
|
|
tracked_task = pending_local_tasks.get(event.call_id)
|
|
if tracked_task is not None and not tracked_task.done():
|
|
tracked_task.cancel()
|
|
|
|
elif isinstance(event, NativeToolCall):
|
|
await _call_hook(
|
|
hooks.on_native_tool_call,
|
|
NativeToolCallCtx(tool_type=event.tool_type, data=event.data),
|
|
)
|
|
|
|
elif isinstance(event, MessageDone):
|
|
if in_reasoning:
|
|
in_reasoning = False
|
|
await _call_hook(
|
|
hooks.on_reasoning_end,
|
|
ReasoningEndCtx(
|
|
reasoning_text=reasoning_text, summary_text=summary_text
|
|
),
|
|
)
|
|
await _call_hook(
|
|
hooks.on_message_end,
|
|
MessageEndCtx(content=event.content),
|
|
)
|
|
message_started = False
|
|
|
|
elif isinstance(event, OutputFileDone):
|
|
await _call_hook(
|
|
hooks.on_file_output,
|
|
FileOutputCtx(
|
|
file_id=event.file_id,
|
|
filename=event.filename,
|
|
content_type=event.content_type,
|
|
),
|
|
)
|
|
|
|
elif isinstance(event, RetryEvent):
|
|
await _call_hook(
|
|
hooks.on_retry,
|
|
RetryCtx(
|
|
source=event.source,
|
|
tool_name=event.tool_name,
|
|
attempt=event.attempt,
|
|
max_attempts=event.max_attempts,
|
|
delay_seconds=event.delay_seconds,
|
|
error=event.error,
|
|
),
|
|
)
|
|
|
|
elif isinstance(event, ErrorEvent):
|
|
await _call_hook(
|
|
hooks.on_server_error,
|
|
ServerErrorCtx(
|
|
source=event.source,
|
|
tool_name=event.tool_name,
|
|
error=event.error,
|
|
),
|
|
)
|
|
|
|
elif isinstance(
|
|
event,
|
|
ResponseCompleted
|
|
| ResponseFailed
|
|
| ResponseIncomplete
|
|
| ResponseCancelled,
|
|
):
|
|
if in_reasoning:
|
|
in_reasoning = False
|
|
await _call_hook(
|
|
hooks.on_reasoning_end,
|
|
ReasoningEndCtx(
|
|
reasoning_text=reasoning_text, summary_text=summary_text
|
|
),
|
|
)
|
|
status = event.response.status
|
|
await _call_hook(
|
|
hooks.on_response_end,
|
|
ResponseEndCtx(response=event.response, status=status),
|
|
)
|
|
|
|
# ── Yield event to consumer ──────────────
|
|
yield event
|
|
|
|
# ── Post-stream: tool loop ───────────────────
|
|
if current_response_id is not None:
|
|
current_prev_id = current_response_id
|
|
|
|
# Filter out calls that already have server-side results.
|
|
pending_client_calls = [
|
|
tc for tc in pending_client_calls if tc.call_id not in completed_call_ids
|
|
]
|
|
|
|
if not pending_client_calls or tool_handler is None:
|
|
break
|
|
|
|
# Execute client-side tools and build results.
|
|
results: list[dict[str, object]] = []
|
|
result_infos: list[ToolResultInfo] = []
|
|
|
|
for tc in pending_client_calls:
|
|
call_info = ToolCallInfo(
|
|
name=tc.name,
|
|
arguments=tc.arguments,
|
|
call_id=tc.call_id,
|
|
agent_name=tc.agent_name,
|
|
response_id=current_response_id or "",
|
|
iteration=iteration,
|
|
)
|
|
try:
|
|
# Sync ``execute`` routes through
|
|
# ``asyncio.to_thread`` so a blocking body
|
|
# doesn't serialize the sync-tool loop
|
|
# (each call would otherwise freeze the
|
|
# event loop for its duration).
|
|
output = await _call_execute_off_loop(tool_handler, call_info)
|
|
except ToolCallDenied as exc:
|
|
output = str(exc)
|
|
|
|
await _call_hook(
|
|
hooks.on_tool_call_end,
|
|
ToolCallEndCtx(
|
|
name=tc.name,
|
|
call_id=tc.call_id,
|
|
agent_name=tc.agent_name,
|
|
output=output,
|
|
),
|
|
)
|
|
yield ToolResult(call_id=tc.call_id, output=output)
|
|
|
|
results.append(
|
|
{
|
|
"type": "function_call_output",
|
|
"call_id": tc.call_id,
|
|
"output": output,
|
|
}
|
|
)
|
|
result_infos.append(
|
|
ToolResultInfo(
|
|
call_id=tc.call_id,
|
|
name=tc.name,
|
|
output=output,
|
|
agent_name=tc.agent_name,
|
|
)
|
|
)
|
|
|
|
await _call_hook(
|
|
hooks.on_tool_results_ready,
|
|
ToolResultsReadyCtx(results=result_infos, iteration=iteration),
|
|
)
|
|
|
|
current_input = results
|
|
iteration += 1
|
|
|
|
async def get(self, response_id: str) -> Response:
|
|
"""Get a response by ID (poll for status).
|
|
|
|
:param response_id: The response/task ID.
|
|
:returns: Current response state.
|
|
"""
|
|
self._deprecation_warn("get")
|
|
resp = await self._http.get(f"{self._base}/v1/responses/{response_id}")
|
|
raise_for_status(resp.status_code, response_body(resp))
|
|
return Response.from_dict(require_json_object(resp, "GET /v1/responses/{response_id}"))
|
|
|
|
async def poll(
|
|
self,
|
|
response_id: str,
|
|
*,
|
|
interval: float = 0.5,
|
|
tool_handler: ToolHandler | None = None,
|
|
) -> Response:
|
|
"""Poll a background response until it reaches a terminal status.
|
|
|
|
If ``tool_handler`` is provided, tunneled tool calls
|
|
(``status: "action_required"``) are executed and PATCHed back.
|
|
|
|
:param response_id: The response/task ID.
|
|
:param interval: Seconds between polls.
|
|
:param tool_handler: Optional client-side tool handler.
|
|
:returns: The terminal response.
|
|
"""
|
|
while True:
|
|
response = await self.get(response_id)
|
|
if response.status in _TERMINAL_STATUSES:
|
|
return response
|
|
# Check for tunneled tool calls needing client execution.
|
|
if tool_handler is not None:
|
|
await self._handle_polling_tool_calls(response_id, response, tool_handler)
|
|
await asyncio.sleep(interval)
|
|
|
|
async def _handle_polling_tool_calls(
|
|
self,
|
|
response_id: str,
|
|
response: Response,
|
|
tool_handler: ToolHandler,
|
|
) -> None:
|
|
"""Execute action_required tool calls found during polling."""
|
|
action_required = [
|
|
item
|
|
for item in response.output
|
|
if isinstance(item, dict)
|
|
and item.get("type") == "function_call"
|
|
and item.get("status") == "action_required"
|
|
]
|
|
if not action_required:
|
|
return
|
|
|
|
tool_results = []
|
|
for fc in action_required:
|
|
name = str(fc.get("name", ""))
|
|
call_id = str(fc.get("call_id", ""))
|
|
args_str = str(fc.get("arguments", "{}"))
|
|
try:
|
|
arguments = json.loads(args_str)
|
|
except json.JSONDecodeError:
|
|
arguments = {}
|
|
|
|
call_info = ToolCallInfo(
|
|
name=name,
|
|
arguments=arguments,
|
|
call_id=call_id,
|
|
agent_name=str(fc.get("model", "")),
|
|
response_id=response_id,
|
|
iteration=0,
|
|
)
|
|
output = tool_handler.execute(call_info)
|
|
if inspect.isawaitable(output):
|
|
output = await output
|
|
tool_results.append({"call_id": call_id, "output": output})
|
|
|
|
if tool_results:
|
|
await self._patch_tool_results(response_id, tool_results)
|
|
|
|
async def _patch_tool_results(
|
|
self,
|
|
response_id: str,
|
|
tool_results: list[dict[str, str]],
|
|
) -> None:
|
|
"""PATCH tool results back to the server."""
|
|
resp = await self._http.patch(
|
|
f"{self._base}/v1/responses/{response_id}",
|
|
json={"tool_results": tool_results},
|
|
timeout=60.0,
|
|
)
|
|
if resp.status_code not in (200, 404, 409):
|
|
_log.warning(
|
|
"PATCH tool results failed (%d): %s",
|
|
resp.status_code,
|
|
resp.text[:200],
|
|
)
|
|
|
|
async def steer(
|
|
self,
|
|
response_id: str,
|
|
input: str,
|
|
*,
|
|
model: str,
|
|
reasoning: dict[str, str] | None = None,
|
|
model_override: str | None = None,
|
|
) -> Response:
|
|
"""Send a steering message to an in-progress response.
|
|
|
|
:param response_id: The in-progress response ID.
|
|
:param input: Steering text.
|
|
:param model: Agent name.
|
|
:param reasoning: Reasoning config if this steer races into a new response.
|
|
:param model_override: LLM model override if this steer races
|
|
into a new response (server ignores it otherwise).
|
|
:returns: The response (same ID if delivered, new ID if agent finished).
|
|
"""
|
|
self._deprecation_warn("steer")
|
|
body: dict[str, object] = {
|
|
"model": model,
|
|
"input": input,
|
|
"previous_response_id": response_id,
|
|
"stream": False,
|
|
"background": True,
|
|
}
|
|
if reasoning is not None:
|
|
body["reasoning"] = reasoning
|
|
if model_override is not None:
|
|
body["model_override"] = model_override
|
|
resp = await self._http.post(
|
|
f"{self._base}/v1/responses",
|
|
json=body,
|
|
timeout=120.0,
|
|
)
|
|
raise_for_status(resp.status_code, response_body(resp))
|
|
return Response.from_dict(require_json_object(resp, "POST /v1/responses"))
|
|
|
|
async def cancel(self, response_id: str) -> Response:
|
|
"""Cancel an in-progress response.
|
|
|
|
:param response_id: The response ID to cancel.
|
|
:returns: The cancelled response.
|
|
"""
|
|
self._deprecation_warn("cancel")
|
|
resp = await self._http.post(
|
|
f"{self._base}/v1/responses/{response_id}/cancel",
|
|
timeout=10.0,
|
|
)
|
|
raise_for_status(resp.status_code, response_body(resp))
|
|
return Response.from_dict(
|
|
require_json_object(resp, "POST /v1/responses/{response_id}/cancel")
|
|
)
|
|
|
|
async def delete(self, response_id: str) -> None:
|
|
"""Delete a response.
|
|
|
|
:param response_id: The response ID to delete.
|
|
"""
|
|
self._deprecation_warn("delete")
|
|
resp = await self._http.delete(
|
|
f"{self._base}/v1/responses/{response_id}",
|
|
)
|
|
if resp.status_code >= 400:
|
|
raise_for_status(resp.status_code, response_body(resp))
|
|
|
|
|
|
# ── Helpers ──────────────────────────────────────────────
|
|
|
|
|
|
def _build_body(
|
|
*,
|
|
model: str,
|
|
input: str | list[dict[str, object]],
|
|
stream: bool,
|
|
background: bool,
|
|
instructions: str | None,
|
|
previous_response_id: str | None,
|
|
tools: list[dict[str, object]] | None,
|
|
reasoning: dict[str, str] | None,
|
|
model_override: str | None = None,
|
|
) -> dict[str, object]:
|
|
"""Build the request body for POST /v1/responses."""
|
|
body: dict[str, object] = {
|
|
"model": model,
|
|
"input": input,
|
|
"stream": stream,
|
|
}
|
|
if background:
|
|
body["background"] = True
|
|
if instructions is not None:
|
|
body["instructions"] = instructions
|
|
if previous_response_id is not None:
|
|
body["previous_response_id"] = previous_response_id
|
|
if tools is not None:
|
|
body["tools"] = tools
|
|
if reasoning is not None:
|
|
body["reasoning"] = reasoning
|
|
if model_override is not None:
|
|
body["model_override"] = model_override
|
|
return body
|
|
|
|
|
|
async def _execute_and_patch(
|
|
http: httpx.AsyncClient,
|
|
base_url: str,
|
|
tool_handler: ToolHandler,
|
|
hooks: StreamHooks,
|
|
tool_call: ToolCall,
|
|
root_response_id: str,
|
|
iteration: int,
|
|
) -> None:
|
|
"""Execute a tunneled sub-agent tool call and PATCH the result back.
|
|
|
|
Runs in the background via ``asyncio.ensure_future`` while the
|
|
SSE stream continues.
|
|
"""
|
|
call_info = ToolCallInfo(
|
|
name=tool_call.name,
|
|
arguments=tool_call.arguments,
|
|
call_id=tool_call.call_id,
|
|
agent_name=tool_call.agent_name,
|
|
response_id=root_response_id,
|
|
iteration=iteration,
|
|
)
|
|
try:
|
|
# Sync ``execute`` goes through asyncio.to_thread so a
|
|
# blocking body doesn't stall the SSE stream or the
|
|
# TUI's render loop. See ``_call_execute_off_loop``.
|
|
output = await _call_execute_off_loop(tool_handler, call_info)
|
|
except ToolCallDenied as exc:
|
|
output = str(exc)
|
|
except Exception:
|
|
_log.exception("Error executing tunneled tool call %s", tool_call.name)
|
|
output = f"Error executing tool: {tool_call.name}"
|
|
|
|
await _call_hook(
|
|
hooks.on_tool_call_end,
|
|
ToolCallEndCtx(
|
|
name=tool_call.name,
|
|
call_id=tool_call.call_id,
|
|
agent_name=tool_call.agent_name,
|
|
output=output,
|
|
),
|
|
)
|
|
|
|
try:
|
|
resp = await http.patch(
|
|
f"{base_url}/v1/responses/{root_response_id}",
|
|
json={"tool_results": [{"call_id": tool_call.call_id, "output": output}]},
|
|
timeout=60.0,
|
|
)
|
|
if resp.status_code not in (200, 404, 409):
|
|
_log.warning(
|
|
"PATCH failed for call_id %s: %s",
|
|
tool_call.call_id,
|
|
resp.text[:200],
|
|
)
|
|
except Exception:
|
|
_log.exception("Error PATCHing tool result for call_id %s", tool_call.call_id)
|
|
|
|
|
|
async def _call_execute_off_loop(
|
|
tool_handler: ToolHandler,
|
|
call_info: ToolCallInfo,
|
|
) -> Any:
|
|
"""
|
|
Invoke ``tool_handler.execute(call_info)`` without blocking
|
|
the event loop.
|
|
|
|
``execute`` may be ``async def`` (returns a coroutine) or
|
|
sync ``def`` (returns a plain value). If async, we await
|
|
the coroutine on the loop; async bodies are already
|
|
loop-cooperative. If sync, we run it via
|
|
:func:`asyncio.to_thread` so blocking calls inside —
|
|
``time.sleep``, file I/O, subprocess, ``requests`` — don't
|
|
stall the loop.
|
|
|
|
Sync-vs-async is detected via
|
|
:func:`inspect.iscoroutinefunction` so we only invoke
|
|
``execute`` once. Without the thread bounce, a single
|
|
sync body with ``time.sleep(5)`` would serialize every
|
|
concurrent tool call AND freeze any render loop (like the
|
|
``omnigent chat`` TUI) sharing the event loop.
|
|
|
|
:param tool_handler: Handler whose ``execute`` will run.
|
|
:param call_info: Call context passed through to the handler.
|
|
:returns: Whatever ``execute`` returns (typically a string).
|
|
"""
|
|
if inspect.iscoroutinefunction(tool_handler.execute):
|
|
return await tool_handler.execute(call_info)
|
|
return await asyncio.to_thread(tool_handler.execute, call_info)
|
|
|
|
|
|
async def _invoke_elicitation_hook(
|
|
hooks: StreamHooks,
|
|
ctx: ElicitationRequestCtx,
|
|
) -> bool:
|
|
"""
|
|
Run the consumer's ``on_elicitation_request`` hook fail-closed.
|
|
|
|
Returns True only when the hook explicitly accepts. A missing
|
|
hook, an exception from the hook, or a falsy return all map to
|
|
False (decline) — the parked workflow must not stall on an
|
|
elicitation no consumer can answer.
|
|
|
|
:param hooks: Stream hooks — the ``on_elicitation_request``
|
|
callable to invoke.
|
|
:param ctx: Context carrying the parsed elicitation, passed to
|
|
the hook verbatim.
|
|
:returns: ``True`` iff the hook explicitly accepted; ``False``
|
|
on missing-hook, exception, or any falsy return.
|
|
"""
|
|
if hooks.on_elicitation_request is None:
|
|
_log.info(
|
|
"No on_elicitation_request hook registered; declining elicitation_id=%s policy=%s",
|
|
ctx.elicitation_id,
|
|
ctx.policy_name,
|
|
)
|
|
return False
|
|
try:
|
|
raw = hooks.on_elicitation_request(ctx)
|
|
if inspect.isawaitable(raw):
|
|
raw = await raw
|
|
return bool(raw)
|
|
except Exception:
|
|
_log.exception(
|
|
"on_elicitation_request hook raised for elicitation_id %s; declining fail-closed",
|
|
ctx.elicitation_id,
|
|
)
|
|
return False
|
|
|
|
|
|
async def _post_elicitation_result(
|
|
http: httpx.AsyncClient,
|
|
base_url: str,
|
|
session_id: str,
|
|
elicitation_id: str,
|
|
accepted: bool,
|
|
) -> None:
|
|
"""
|
|
POST a verdict to one elicitation's dedicated resolve URL.
|
|
|
|
URL-based elicitation: the verdict goes to
|
|
``POST /v1/sessions/{id}/elicitations/{eid}/resolve`` with the
|
|
MCP ``ElicitResult`` body (``{"action": "accept"}`` or
|
|
``{"action": "decline"}``) — the elicitation id rides in the URL
|
|
path. ``content`` is omitted for binary accept/decline (future
|
|
richer hooks would populate it). This matches the REPL and web
|
|
clients; the server routes all of them through the same
|
|
``_resolve_elicitation``.
|
|
|
|
Status codes the server returns (202/404/409) are all benign —
|
|
the verdict is either applied or moot. Anything else gets a
|
|
warning log; the call still returns normally because the
|
|
background task is fire-and-forget.
|
|
|
|
:param http: Shared HTTPX client.
|
|
:param base_url: Server base URL.
|
|
:param session_id: Session/conversation id that emitted the
|
|
elicitation. The resolve URL is scoped to this session so
|
|
the server can apply normal session authorization.
|
|
:param elicitation_id: Server-assigned elicitation id.
|
|
:param accepted: Hook verdict — ``True`` → ``"accept"``,
|
|
``False`` → ``"decline"``.
|
|
"""
|
|
if not session_id:
|
|
_log.warning(
|
|
"Cannot resolve elicitation_id %s: missing session id",
|
|
elicitation_id,
|
|
)
|
|
return
|
|
body = {"action": "accept" if accepted else "decline"}
|
|
try:
|
|
resp = await http.post(
|
|
f"{base_url}/v1/sessions/{session_id}/elicitations/{elicitation_id}/resolve",
|
|
json=body,
|
|
timeout=60.0,
|
|
)
|
|
# 202 = delivered, 404 = session/parked workflow already gone
|
|
# (cancel race), 409 = already completed. All three are
|
|
# benign: the verdict is either applied or moot.
|
|
if resp.status_code not in (202, 404, 409):
|
|
_log.warning(
|
|
"Resolve elicitation_id %s failed: %s",
|
|
elicitation_id,
|
|
resp.text[:200],
|
|
)
|
|
except Exception:
|
|
_log.exception(
|
|
"Error POSTing approval event for elicitation_id %s",
|
|
elicitation_id,
|
|
)
|
|
|
|
|
|
async def _handle_elicitation_request(
|
|
http: httpx.AsyncClient,
|
|
base_url: str,
|
|
hooks: StreamHooks,
|
|
event: ElicitationRequest,
|
|
response_id: str,
|
|
session_id: str,
|
|
) -> None:
|
|
"""
|
|
Route an elicitation through the hook and POST the verdict.
|
|
|
|
Runs on a background task so the SSE stream continues to drain
|
|
while the user is deciding. Two-step: invoke the hook
|
|
(:func:`_invoke_elicitation_hook`), POST the approval event
|
|
(:func:`_post_elicitation_result`).
|
|
|
|
:param http: Shared HTTPX client.
|
|
:param base_url: Server base URL.
|
|
:param hooks: Stream hooks — the ``on_elicitation_request``
|
|
hook gets the decision.
|
|
:param event: The parsed elicitation request.
|
|
:param response_id: Root response id (the parked workflow);
|
|
carried into the ctx for the hook's audit/logging only.
|
|
:param session_id: Session/conversation id that emitted the
|
|
elicitation. The verdict is posted back as a session
|
|
``approval`` event.
|
|
"""
|
|
ctx = ElicitationRequestCtx(
|
|
elicitation_id=event.elicitation_id,
|
|
message=event.message,
|
|
requested_schema=event.requested_schema,
|
|
mode=event.mode,
|
|
phase=event.phase,
|
|
policy_name=event.policy_name,
|
|
content_preview=event.content_preview,
|
|
response_id=response_id,
|
|
url=event.url,
|
|
)
|
|
accepted = await _invoke_elicitation_hook(hooks, ctx)
|
|
await _post_elicitation_result(http, base_url, session_id, event.elicitation_id, accepted)
|