in src/agents/tool.py [0:0]
def _create_function_tool(the_func: ToolFunction[...]) -> FunctionTool:
schema = function_schema(
func=the_func,
name_override=name_override,
description_override=description_override,
docstring_style=docstring_style,
use_docstring_info=use_docstring_info,
strict_json_schema=strict_mode,
)
async def _on_invoke_tool_impl(ctx: RunContextWrapper[Any], input: str) -> Any:
try:
json_data: dict[str, Any] = json.loads(input) if input else {}
except Exception as e:
if _debug.DONT_LOG_TOOL_DATA:
logger.debug(f"Invalid JSON input for tool {schema.name}")
else:
logger.debug(f"Invalid JSON input for tool {schema.name}: {input}")
raise ModelBehaviorError(
f"Invalid JSON input for tool {schema.name}: {input}"
) from e
if _debug.DONT_LOG_TOOL_DATA:
logger.debug(f"Invoking tool {schema.name}")
else:
logger.debug(f"Invoking tool {schema.name} with input {input}")
try:
parsed = (
schema.params_pydantic_model(**json_data)
if json_data
else schema.params_pydantic_model()
)
except ValidationError as e:
raise ModelBehaviorError(f"Invalid JSON input for tool {schema.name}: {e}") from e
args, kwargs_dict = schema.to_call_args(parsed)
if not _debug.DONT_LOG_TOOL_DATA:
logger.debug(f"Tool call args: {args}, kwargs: {kwargs_dict}")
if inspect.iscoroutinefunction(the_func):
if schema.takes_context:
result = await the_func(ctx, *args, **kwargs_dict)
else:
result = await the_func(*args, **kwargs_dict)
else:
if schema.takes_context:
result = the_func(ctx, *args, **kwargs_dict)
else:
result = the_func(*args, **kwargs_dict)
if _debug.DONT_LOG_TOOL_DATA:
logger.debug(f"Tool {schema.name} completed.")
else:
logger.debug(f"Tool {schema.name} returned {result}")
return result
async def _on_invoke_tool(ctx: RunContextWrapper[Any], input: str) -> Any:
try:
return await _on_invoke_tool_impl(ctx, input)
except Exception as e:
if failure_error_function is None:
raise
result = failure_error_function(ctx, e)
if inspect.isawaitable(result):
return await result
_error_tracing.attach_error_to_current_span(
SpanError(
message="Error running tool (non-fatal)",
data={
"tool_name": schema.name,
"error": str(e),
},
)
)
return result
return FunctionTool(
name=schema.name,
description=schema.description or "",
params_json_schema=schema.params_json_schema,
on_invoke_tool=_on_invoke_tool,
strict_json_schema=strict_mode,
)