Bases: Toolkit
Toolkit wrapper that enforces OpenAI-compatible strict schemas.
Key behavior: - When registering a callable, we create an agno Function with strict=True. - When registering an agno Function (from @tool decorator), we delegate to the base Toolkit.register().
This keeps the override signature compatible with agno (fixing the “red” method warning) while preserving strict behavior for normal callables.
Orchestration auto-load: - On toolkit construction, we attempt to load once the orchestration singleton from get_orchestration_tools(). - We also auto-register a small set of orchestration functions into every toolkit instance, so that planning/orchestration tools are available even if the user only adds one toolkit.
Tool names are prefixed with orchestrator_ to avoid collisions.
Source code in src/enhancedtoolkits/base.py
| def __init__(self, *args: Any, **kwargs: Any):
super().__init__(*args, **kwargs)
# Skip injection for the orchestration toolkit itself.
if getattr(self, "_disable_auto_orchestrator", False):
return
try:
# Local import to avoid circular import at module load time.
from .orchestration import get_orchestration_tools
orchestrator = get_orchestration_tools()
# Expose for direct Python usage.
self.orchestrator = orchestrator
# Auto-inject orchestration functions (available when using any toolkit).
injections = [
("orchestrator_create_plan", orchestrator.create_plan),
("orchestrator_add_task", orchestrator.add_task),
(
"orchestrator_update_task_status",
orchestrator.update_task_status,
),
("orchestrator_next_actions", orchestrator.next_actions),
(
"orchestrator_summarize_progress",
orchestrator.summarize_progress,
),
("orchestrator_reset_plan", orchestrator.reset_plan),
]
for tool_name, fn in injections:
# Avoid re-registering if a derived toolkit already defines it.
if tool_name in self.functions:
continue
self.register(fn, name=tool_name)
except Exception as exc: # pylint: disable=broad-exception-caught
logger.warning("Failed to auto-load orchestration tools: %s", exc)
|
register(function: Union[Callable[..., Any], Function], name: Optional[str] = None) -> None
Register a callable or agno Function with OpenAI-compatible strictness.
Source code in src/enhancedtoolkits/base.py
| def register(
self,
function: Union[Callable[..., Any], Function],
name: Optional[str] = None,
) -> None:
"""Register a callable or agno `Function` with OpenAI-compatible strictness."""
try:
# agno supports registering already-decorated Function objects. Keep that behavior.
if isinstance(function, Function):
tool_name = name or function.name
super().register(function, name=name)
# Validate OpenAI compatibility only if the function was actually registered.
registered = self.functions.get(tool_name)
if (
registered is not None
and registered.entrypoint is not None
):
self._validate_openai_compatibility(
registered, registered.entrypoint
)
return
tool_name = name or function.__name__
if (
self.include_tools is not None
and tool_name not in self.include_tools
):
return
if (
self.exclude_tools is not None
and tool_name in self.exclude_tools
):
return
# Create Function directly with strict=True
f = Function.from_callable(function, name=tool_name, strict=True)
# Validate OpenAI compatibility
self._validate_openai_compatibility(f, function)
# Set all the necessary properties
f.cache_results = self.cache_results
f.cache_dir = self.cache_dir
f.cache_ttl = self.cache_ttl
f.requires_confirmation = (
tool_name in self.requires_confirmation_tools
)
f.external_execution = (
tool_name in self.external_execution_required_tools
)
f.stop_after_tool_call = (
tool_name in self.stop_after_tool_call_tools
)
f.show_result = tool_name in self.show_result_tools
# Add to functions dictionary
self.functions[f.name] = f
log_debug(
f"Function: {f.name} registered with {self.name} (strict=True, OpenAI compatible)"
)
except Exception:
func_name = (
function.name
if isinstance(function, Function)
else function.__name__
)
logger.warning("Failed to create Function for: %s", func_name)
raise
|