Skip to content

Commit 95d9645

Browse files
apcha-oaiapcha-oai
authored
feat(agents): observe local tool failures (#4028)
## Summary Observe local tool failures without sending application exceptions to the model. Previously, logging inside a handler missed argument-validation failures before invocation and serialization failures after it returned: ```python with client.beta.agents.sessions.stream( SESSION_ID, input="Find catalog item A123.", tool_handlers=handlers, ) as stream: stream.until_done() ``` Now attach an optional observer alongside the handlers: ```python def report_tool_error(failure): logger.error( "Tool %s failed during %s (call %s)", failure.tool_name, failure.stage, failure.call_id, ) with client.beta.agents.sessions.stream( SESSION_ID, input="Find catalog item A123.", tool_handlers=handlers, on_tool_error=report_tool_error, ) as stream: stream.until_done() ``` The beta `AgentToolError` carries the original local exception, failure stage, and call identifiers. `AsyncOpenAI` also accepts awaitable observers. Ordinary observer exceptions do not prevent submission of the sanitized tool failure; cancellation and API/stream errors retain their existing propagation. No automatic logging is added.
1 parent becc1d2 commit 95d9645

9 files changed

Lines changed: 399 additions & 25 deletions

File tree

‎helpers.md‎

Lines changed: 27 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -605,6 +605,33 @@ lookup = pydantic_function_tool(
605605
606606
Callbacks can be async when used with `AsyncOpenAI`. Existing dictionary handlers still work.
607607
608+
### Observing local tool failures
609+
610+
Use `on_tool_error` to log or monitor failures in argument validation, handler
611+
execution, or output serialization. The callback receives an `AgentToolError`
612+
from `openai.lib.beta.agents`; its original exception stays local. The model
613+
still receives the generic tool failure, and the SDK does not log automatically.
614+
Exception messages and tracebacks may contain sensitive data; apply your
615+
application’s redaction policy before logging them.
616+
617+
```python
618+
def report_tool_error(failure):
619+
logger.error(
620+
"Tool %s failed during %s (call %s)",
621+
failure.tool_name, failure.stage, failure.call_id,
622+
)
623+
624+
with client.beta.agents.sessions.stream(
625+
SESSION_ID, input="Find catalog item A123.", tool_handlers=handlers,
626+
on_tool_error=report_tool_error,
627+
) as stream:
628+
stream.until_done()
629+
```
630+
631+
`AsyncOpenAI` also accepts an async observer. Ordinary observer exceptions are
632+
ignored so the original tool failure can still be submitted; cancellation is
633+
not suppressed. Submission errors propagate normally without invoking this callback.
634+
608635
### Typed Agents output (beta)
609636

610637
Pass a Pydantic model (or a Pydantic v2 dataclass) to generate the Agents output

‎src/openai/lib/beta/agents/__init__.py‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -25,3 +25,4 @@
2525
AgentResultArtifacts as AgentResultArtifacts,
2626
AsyncAgentResultArtifacts as AsyncAgentResultArtifacts,
2727
)
28+
from ._tool_error import AgentToolError as AgentToolError

‎src/openai/lib/beta/agents/_tool_error.py‎

Lines changed: 29 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,29 @@
1+
from __future__ import annotations
2+
3+
from typing import Callable, Awaitable
4+
from dataclasses import dataclass
5+
from typing_extensions import Literal, TypeAlias
6+
7+
ToolErrorStage: TypeAlias = Literal["arguments", "execution", "output"]
8+
9+
10+
@dataclass(frozen=True)
11+
class AgentToolError:
12+
"""Local diagnostic passed to ``sessions.stream(on_tool_error=...)``.
13+
14+
Use alongside ``tool_handlers`` to log or monitor argument validation,
15+
execution, and output serialization failures. ``error`` is the original
16+
local exception; it is never sent to the model. API and stream errors
17+
propagate normally without invoking the callback.
18+
"""
19+
20+
error: Exception
21+
tool_name: str
22+
session_id: str
23+
turn_id: str
24+
call_id: str
25+
stage: ToolErrorStage
26+
27+
28+
ToolErrorHandler: TypeAlias = Callable[[AgentToolError], None]
29+
AsyncToolErrorHandler: TypeAlias = Callable[[AgentToolError], None | Awaitable[None]]

‎src/openai/lib/beta/agents/_tools.py‎

Lines changed: 37 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -15,6 +15,7 @@
1515
from ...._utils import is_dict
1616
from ...._compat import PYDANTIC_V1, model_dump, model_json, model_parse
1717
from ..._pydantic import resolve_ref
18+
from ._tool_error import ToolErrorStage
1819
from ...streaming.agents._types import ToolOutput
1920
from ....types.beta.agent_tool_param import AgentToolConfigParamFunction
2021

@@ -60,10 +61,9 @@ def __init__(
6061
"parameters": parameters,
6162
}
6263

63-
def invoke(arguments: dict[str, Any]) -> _OutputT:
64-
return handler(model_parse(model, arguments))
65-
66-
self._invoke = invoke
64+
self._model = model
65+
self._handler = handler
66+
self._convert_output = False
6767

6868
@property
6969
def name(self) -> str:
@@ -75,7 +75,31 @@ def definition(self) -> AgentToolConfigParamFunction:
7575
return deepcopy(self._definition)
7676

7777
def __call__(self, arguments: dict[str, Any]) -> _OutputT:
78-
return self._invoke(arguments)
78+
return self._call(arguments)
79+
80+
def _call(self, arguments: dict[str, Any], set_stage: Callable[[ToolErrorStage], None] | None = None) -> _OutputT:
81+
if set_stage is not None:
82+
set_stage("arguments")
83+
parsed = model_parse(self._model, arguments)
84+
if set_stage is not None:
85+
set_stage("execution")
86+
if self._convert_output and inspect.iscoroutinefunction(self._handler):
87+
88+
async def invoke_async() -> ToolOutput:
89+
output = await cast(Awaitable[object], self._handler(parsed))
90+
if set_stage is not None:
91+
set_stage("output")
92+
return _tool_output(output)
93+
94+
return cast(_OutputT, invoke_async())
95+
output = self._handler(parsed)
96+
if not self._convert_output:
97+
return output
98+
if inspect.isawaitable(output):
99+
return cast(_OutputT, _AwaitableToolOutput(output, set_stage))
100+
if set_stage is not None:
101+
set_stage("output")
102+
return cast(_OutputT, _tool_output(output))
79103

80104

81105
def _tool_output(output: object) -> ToolOutput:
@@ -103,11 +127,14 @@ def _tool_output(output: object) -> ToolOutput:
103127

104128

105129
class _AwaitableToolOutput:
106-
def __init__(self, output: Awaitable[object]) -> None:
130+
def __init__(self, output: Awaitable[object], set_stage: Callable[[ToolErrorStage], None] | None = None) -> None:
107131
self._output = output
132+
self._set_stage = set_stage
108133

109134
def __await__(self) -> Generator[Any, None, ToolOutput]:
110135
output = yield from self._output.__await__()
136+
if self._set_stage is not None:
137+
self._set_stage("output")
111138
return _tool_output(output)
112139

113140
def close(self) -> None:
@@ -152,21 +179,14 @@ def pydantic_function_tool(
152179
(such as union models) are unsupported; wrap a union in an ordinary model field.
153180
"""
154181

155-
async def invoke_async(arguments: _ModelT) -> ToolOutput:
156-
return _tool_output(await cast(Awaitable[object], handler(arguments)))
157-
158-
def invoke(arguments: _ModelT) -> ToolOutput | Awaitable[ToolOutput]:
159-
output = handler(arguments)
160-
if inspect.isawaitable(output):
161-
return _AwaitableToolOutput(output)
162-
return _tool_output(output)
163-
164-
return FunctionTool(
182+
tool = FunctionTool[Any](
165183
model,
166-
invoke_async if inspect.iscoroutinefunction(handler) else invoke,
184+
cast(Callable[[_ModelT], Any], handler),
167185
name=model.__name__ if name is None else name,
168186
description=(model.__doc__ or "") if description is None else description,
169187
)
188+
tool._convert_output = True
189+
return tool
170190

171191

172192
class _FunctionToolDecorator(Protocol):

‎src/openai/lib/streaming/agents/_streams.py‎

Lines changed: 31 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -10,7 +10,7 @@
1010

1111
import httpx2
1212

13-
from ._tools import arguments, failed_event, result_event, is_pending_call_race
13+
from ._tools import ToolInvocation, failed_event, result_event, is_pending_call_race
1414
from ._types import ToolHandler, AsyncToolHandler
1515
from ...._types import Omit, Headers, NotGiven, omit, not_given
1616
from ...._streaming import Stream, AsyncStream
@@ -22,6 +22,7 @@
2222
AgentOutputParseError,
2323
AgentTurnResultCollection,
2424
)
25+
from ...beta.agents._tool_error import ToolErrorHandler, AsyncToolErrorHandler
2526
from ....types.beta.agent_session_event import AgentSessionEvent
2627
from ....types.beta.agent_function_call_item import AgentFunctionCallItem
2728
from ....types.beta.agent_session_input_param import (
@@ -141,6 +142,7 @@ def __init__(
141142
input: str | Iterable[AgentSessionInputMessageParam],
142143
output_type: type[OutputT] | None = None,
143144
tool_handlers: Mapping[str, ToolHandler] | None = None,
145+
on_tool_error: ToolErrorHandler | None = None,
144146
idempotency_key: str | Omit = omit,
145147
extra_headers: Headers | None = None,
146148
timeout: float | httpx2.Timeout | None | NotGiven = not_given,
@@ -149,6 +151,7 @@ def __init__(
149151
self._session_id = session_id
150152
self._input = _input_event(input)
151153
self._handlers = dict(tool_handlers or {})
154+
self._on_tool_error = on_tool_error
152155
self._idempotency_key = _input_key(idempotency_key, extra_headers)
153156
self._options = _request_options({"extra_headers": extra_headers, "timeout": timeout})
154157
self._state = _TurnState()
@@ -253,16 +256,27 @@ def _iterate(self) -> Iterator[AgentSessionEvent]:
253256
if terminal or self._closed:
254257
return
255258
if call is not None and handler is not None:
259+
invocation = ToolInvocation(self._session_id, call)
256260
try:
257-
output: Any = handler(arguments(call))
261+
output: Any = invocation.invoke(handler)
258262
if inspect.isawaitable(output):
259263
close = getattr(output, "close", None)
260264
if callable(close):
261265
close()
262266
raise TypeError("Async tool handlers require AsyncOpenAI")
267+
invocation.stage = "output"
263268
result = result_event(call, output)
264-
except Exception:
269+
except Exception as error:
265270
result = failed_event(call)
271+
if self._on_tool_error is not None:
272+
try:
273+
notification: Any = self._on_tool_error(invocation.failure(error))
274+
# A sync stream cannot await an async observer.
275+
if inspect.iscoroutine(notification):
276+
notification.close()
277+
except Exception:
278+
# Observer failures must not change tool submission or the turn result.
279+
pass
266280
self._submit_result(result)
267281
raise RuntimeError("Session event stream ended before the turn reached idle or failed")
268282
except Exception as error:
@@ -304,6 +318,7 @@ def __init__(
304318
input: str | Iterable[AgentSessionInputMessageParam],
305319
output_type: type[OutputT] | None = None,
306320
tool_handlers: Mapping[str, AsyncToolHandler] | None = None,
321+
on_tool_error: AsyncToolErrorHandler | None = None,
307322
idempotency_key: str | Omit = omit,
308323
extra_headers: Headers | None = None,
309324
timeout: float | httpx2.Timeout | None | NotGiven = not_given,
@@ -312,6 +327,7 @@ def __init__(
312327
self._session_id = session_id
313328
self._input = _input_event(input)
314329
self._handlers = dict(tool_handlers or {})
330+
self._on_tool_error = on_tool_error
315331
self._idempotency_key = _input_key(idempotency_key, extra_headers)
316332
self._options = _request_options({"extra_headers": extra_headers, "timeout": timeout})
317333
self._state = _TurnState()
@@ -416,13 +432,23 @@ async def _iterate(self) -> AsyncIterator[AgentSessionEvent]:
416432
if terminal or self._closed:
417433
return
418434
if call is not None and handler is not None:
435+
invocation = ToolInvocation(self._session_id, call)
419436
try:
420-
output = handler(arguments(call))
437+
output = invocation.invoke(handler)
421438
if inspect.isawaitable(output):
422439
output = await output
440+
invocation.stage = "output"
423441
result = result_event(call, output)
424-
except Exception:
442+
except Exception as error:
425443
result = failed_event(call)
444+
if self._on_tool_error is not None:
445+
try:
446+
notification: Any = self._on_tool_error(invocation.failure(error))
447+
if inspect.isawaitable(notification):
448+
await notification
449+
except Exception:
450+
# Observer failures must not change tool submission or the turn result.
451+
pass
426452
await self._submit_result(result)
427453
raise RuntimeError("Session event stream ended before the turn reached idle or failed")
428454
except Exception as error:

‎src/openai/lib/streaming/agents/_tools.py‎

Lines changed: 39 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -4,8 +4,11 @@
44
from copy import deepcopy
55
from typing import Any, Dict, Mapping, cast
66

7-
from ._types import ToolOutput
7+
from ._types import ToolOutput, ToolHandler, AsyncToolHandler
8+
from ...._utils import maybe_transform
89
from ...._exceptions import BadRequestError
10+
from ...._utils._json import openapi_dumps
11+
from ...beta.agents._tool_error import AgentToolError, ToolErrorStage
912
from ....types.beta.agent_function_call_item import AgentFunctionCallItem
1013
from ....types.beta.agent_session_input_param import SessionInputParamAgentSessionInputToolResult
1114
from ....types.beta.agent_function_call_output_param import AgentFunctionCallOutputParam
@@ -24,6 +27,9 @@ def result_event(call: AgentFunctionCallItem, output: ToolOutput) -> SessionInpu
2427
# Materialize iterators here so serialization failures produce a failed tool result.
2528
if output is not None and not isinstance(output, str):
2629
output = list(output)
30+
# Use the same content normalization as events.create before checking JSON encoding.
31+
output = maybe_transform(output, AgentFunctionCallOutputParam)
32+
openapi_dumps(output)
2733
return {
2834
"type": "agent.session.input.tool_result",
2935
"turn_id": call.turn_id,
@@ -52,3 +58,35 @@ def is_pending_call_race(error: BadRequestError, call_id: str) -> bool:
5258
return (
5359
body.get("code") == "invalid_request_error" and body.get("message") == f"Unknown pending tool call: {call_id}"
5460
)
61+
62+
63+
class ToolInvocation:
64+
def __init__(self, session_id: str, call: AgentFunctionCallItem) -> None:
65+
self.session_id = session_id
66+
self.call = call
67+
self.stage: ToolErrorStage = "arguments"
68+
69+
def set_stage(self, stage: ToolErrorStage) -> None:
70+
self.stage = stage
71+
72+
def invoke(self, handler: ToolHandler | AsyncToolHandler) -> Any:
73+
from ...beta.agents._tools import FunctionTool
74+
75+
parsed = arguments(self.call)
76+
self.stage = "execution"
77+
if isinstance(handler, FunctionTool):
78+
tool = cast(FunctionTool[Any], handler)
79+
tool_type = cast(type[FunctionTool[Any]], FunctionTool)
80+
if type(tool).__call__ is tool_type.__call__:
81+
return tool._call(parsed, self.set_stage)
82+
return cast(AsyncToolHandler, handler)(parsed)
83+
84+
def failure(self, error: Exception) -> AgentToolError:
85+
return AgentToolError(
86+
error=error,
87+
tool_name=self.call.name,
88+
session_id=self.session_id,
89+
turn_id=self.call.turn_id,
90+
call_id=self.call.call_id,
91+
stage=self.stage,
92+
)

‎src/openai/resources/beta/agents/sessions/sessions.py‎

Lines changed: 13 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -69,6 +69,7 @@
6969
from .....lib.beta.agents._output import bind_output_type, with_output_schema
7070
from .....lib.beta.agents._result import OutputT
7171
from .....types.beta.agent_session import AgentSession
72+
from .....lib.beta.agents._tool_error import ToolErrorHandler, AsyncToolErrorHandler
7273
from .....types.beta.environment_param import EnvironmentParam
7374
from .....types.beta.agent_session_deleted import AgentSessionDeleted
7475
from .....types.beta.agent_session_input_message_param import AgentSessionInputMessageParam
@@ -84,6 +85,7 @@ def stream(
8485
input: str | Iterable[AgentSessionInputMessageParam],
8586
output_type: type[OutputT] | None = None,
8687
tool_handlers: Mapping[str, ToolHandler] | None = None,
88+
on_tool_error: ToolErrorHandler | None = None,
8789
idempotency_key: str | Omit = omit,
8890
extra_headers: Headers | None = None,
8991
timeout: float | httpx2.Timeout | None | NotGiven = not_given,
@@ -92,13 +94,17 @@ def stream(
9294
9395
Use as a context manager. Only one caller may submit input to the session
9496
while this helper runs. Optional tool handlers receive an arguments dict;
95-
their results are submitted automatically. See AgentSessionStream for details.
97+
their results are submitted automatically. Use on_tool_error alongside
98+
tool_handlers to log or monitor local argument, handler, and output failures.
99+
The callback does not receive API or stream errors. Ordinary observer
100+
exceptions are ignored. See AgentSessionStream for details.
96101
"""
97102
return AgentSessionStream(
98103
self,
99104
session_id,
100105
input=input,
101106
tool_handlers=tool_handlers,
107+
on_tool_error=on_tool_error,
102108
output_type=output_type,
103109
idempotency_key=idempotency_key,
104110
extra_headers=extra_headers,
@@ -574,6 +580,7 @@ def stream(
574580
input: str | Iterable[AgentSessionInputMessageParam],
575581
output_type: type[OutputT] | None = None,
576582
tool_handlers: Mapping[str, AsyncToolHandler] | None = None,
583+
on_tool_error: AsyncToolErrorHandler | None = None,
577584
idempotency_key: str | Omit = omit,
578585
extra_headers: Headers | None = None,
579586
timeout: float | httpx2.Timeout | None | NotGiven = not_given,
@@ -582,13 +589,17 @@ def stream(
582589
583590
Use as an async context manager. Only one caller may submit input to the session
584591
while this helper runs. Optional tool handlers receive an arguments dict;
585-
their results are submitted automatically. See AsyncAgentSessionStream for details.
592+
their results are submitted automatically. Use on_tool_error alongside
593+
tool_handlers to log or monitor local argument, handler, and output failures.
594+
The callback may be async; it does not receive API or stream errors.
595+
Ordinary observer exceptions are ignored. See AsyncAgentSessionStream for details.
586596
"""
587597
return AsyncAgentSessionStream(
588598
self,
589599
session_id,
590600
input=input,
591601
tool_handlers=tool_handlers,
602+
on_tool_error=on_tool_error,
592603
output_type=output_type,
593604
idempotency_key=idempotency_key,
594605
extra_headers=extra_headers,

0 commit comments

Comments
 (0)