orxtra v0.13.0 /tool.src.orxtra.tool._pipeline
Updated
On this page

wrap_tool_with_pipeline layers active-task checks, secret substitution, transient-retry execution, secret scrubbing, mutation tracking, and trace callbacks around a Tool; compose bypasses the pipeline for tool-calling-tool.

#tool.src.orxtra.tool._pipeline

#tool.src.orxtra.tool._pipeline

#wrap_tool_with_pipeline

python
def wrap_tool_with_pipeline(tool: Tool, scheduler_check: Callable[[str], UUID], secret_registry: SecretRegistry | None, trace_callback: Callable[..., Any] | None, session_id: str, is_start_task: bool=False, is_file_mutation: bool=False, mutation_tracker: dict[str, set[str]] | None=None, result_appendix: Callable[[str], str | None] | None=None) -> Tool

Return a new Tool with the same schema but a wrapped execute that runs the full pipeline: active-task check, secret substitution, execution, secret scrubbing, mutation tracking, and trace callback.

#wrap_tools_for_session

python
def wrap_tools_for_session(tools: list[Tool], scheduler_check: Callable[[str], UUID], secret_registry: SecretRegistry | None, trace_callback: Callable[..., Any] | None, session_id: str, mutation_tracker: dict[str, set[str]] | None=None, result_appendix: Callable[[str], str | None] | None=None) -> list[Tool]

Wrap all tools in the list with the execution pipeline.

#compose

python
async def compose(tool: Tool, args: dict[str, Any]) -> ToolOutput[Any]

Call a tool's raw execute, bypassing the pipeline.

Use when tool A calls tool B as an implementation detail. B's execution is not traced, not scrubbed, not mutation-tracked. Attribution goes to the outer tool A.

Search