Tool generation helpers — given an EntitySchema, produce plain-function CRUD tools.
ADK auto-generates tool schemas from function name + docstring + type hints,
so each tool is a simple callable (no wrapper dataclass needed).
generate_crud_tools(
entity: EntitySchema,
) -> list[Callable[..., Any]]
Generate CRUD + semantic search tool functions for an entity.
Each returned function has __name__ set to <entity>_<operation>
and a docstring describing the operation — this is all ADK needs to
build the tool schema automatically.
Source code in libs/ninja-agents/src/ninja_agents/tools.py
| def generate_crud_tools(entity: EntitySchema) -> list[Callable[..., Any]]:
"""Generate CRUD + semantic search tool functions for an entity.
Each returned function has ``__name__`` set to ``<entity>_<operation>``
and a docstring describing the operation — this is all ADK needs to
build the tool schema automatically.
"""
tools: list[Callable[..., Any]] = []
for operation, desc_template in _CRUD_OPERATIONS:
description = desc_template.format(entity=entity.name)
tools.append(_make_tool(entity.name, operation, description))
return tools
|
invoke_tool(
tool: Callable[..., Any],
span: AgentSpan | None = None,
**kwargs: Any,
) -> Any
Invoke a tool function, optionally recording the call in a trace span.
Source code in libs/ninja-agents/src/ninja_agents/tools.py
| def invoke_tool(
tool: Callable[..., Any],
span: AgentSpan | None = None,
**kwargs: Any,
) -> Any:
"""Invoke a tool function, optionally recording the call in a trace span."""
start = time.monotonic()
error: str | None = None
success = True
try:
result = tool(**kwargs)
except Exception as exc:
error = str(exc)
success = False
raise
else:
return result
finally:
duration_ms = (time.monotonic() - start) * 1000
if span is not None:
span.record_tool_call(
tool_name=getattr(tool, "__name__", "unknown"),
input_summary=str(kwargs)[:200],
output_summary="" if not success else str(result)[:200], # type: ignore[possibly-undefined]
duration_ms=duration_ms,
success=success,
error=error,
)
|