#313
Add a `wrap` method to `BoundToolCall` for customizing tools
The wrapper with be run inside of the exception handling for the tool, and can then call the innards.
Activity
4
anbuzin1mo agoLet me think about this a little bit more. This feels like it could either be more general or get replaced by some non-API pattern
anbuzin24d agoAs an alternative to adding a new method, we could require users to rely on a custom loop instead.
class UltraServerlessAgent(ai.Agent):
async def loop(self, context: ai.Context) -> AsyncGenerator[ai.events.AgentEvent]:
while context.keep_running():
async with (
ai.experimental_telemetry.span(ai.experimental_telemetry.LoopTurnSpanData()),
ai.stream(context=context) as stream,
ai.ToolRunner() as tr,
):
async for event in ai.util.merge(stream, tr.events()):
yield event
if isinstance(event, ai.events.ToolEnd):
tr.schedule(functools.partial(run_remotely, event.tool_call))
context.add(stream.message)
context.add(tr.get_tool_message())
async def run_remotely(part: ai.messages.ToolCallPart) -> ai.events.ToolCallResult:
# On resume, calls that already finished carry their result.
if part.cached_result is not None:
return ai.tool_result(part.cached_result)
try:
value = await ai.hook(
f"eval_{part.tool_call_id}",
payload=Value,
metadata={"part": part.model_dump(mode="json")},
tool_call_id=part.tool_call_id,
)
except ai.HookDeferredException as e:
return ai.deferred_tool_result(
e.hook, tool_call_id=part.tool_call_id, tool_name=part.tool_name
)
if value.result is not None:
return value.result
return ai.tool_result(
tool_call_id=part.tool_call_id,
tool_name=part.tool_name,
result=value.error_message,
is_error=True,
)
The latest updates on your projects. Learn more about Vercel for GitHub.