On this page
Builds a Tool from an HttpExecution definition: substitutes {{secret:NAME}} placeholders at call time, interpolates {param} in the URL, issues the httpx request, and validates the response against the output schema.
#tool.src.orxtra.tool._data_tool_http
#tool.src.orxtra.tool._data_tool_http
Factory for building Tool instances from HttpExecution definitions.
Takes a DataToolDefinition with an HttpExecution config and builds a concrete Tool whose execute function:
- Validates agent-supplied args against the param schema.
- Substitutes
{{secret:NAME}}placeholders in URL, headers, and
body_template at CALL TIME (real values never enter the Tool object or LLM-visible text).
- Interpolates
{param_name}in the URL with URL-encoded arg
values, validating against param patterns when defined.
- Makes the HTTP request via httpx.
- Validates the response body against the output schema (if defined).
- Returns a ToolOutput with validated/projected data and rendered text.
#_build_json_schema_params
def _build_json_schema_params(params: dict[str, ParamDef]) -> dict[str, Any]Build a JSON Schema parameters dict from ParamDef entries.
Delegates to the shared implementation in _data_tool_shared.
#_substitute_secrets
def _substitute_secrets(text: str, secret_registry: SecretRegistry | None) -> strReplace {{secret:NAME}} placeholders with real values.
Hard error if placeholders exist but no registry is provided.
#_interpolate_url
def _interpolate_url(url_template: str, args: dict[str, Any], params: dict[str, ParamDef]) -> strSubstitute {param_name} in the URL with URL-encoded arg values.
Validates interpolated param values against their pattern if one is defined. Raises ToolError on missing params or pattern mismatches.
#_validate_output_schema
def _validate_output_schema(response_data: Any, schema: dict[str, Any]) -> NoneValidate response data against the output JSON Schema.
Delegates to the shared implementation in _data_tool_shared.
#_validate_args
def _validate_args(args: dict[str, Any], params: dict[str, ParamDef]) -> NoneValidate that required params are present and types match.
Delegates to the shared implementation in _data_tool_shared.
#_apply_secret_substitution
def _apply_secret_substitution(url_template: str, header_template: dict[str, str] | None, body_tmpl: str | None, secret_registry: SecretRegistry | None) -> tuple[str, dict[str, str] | None, str | None]Apply secret substitution to URL, headers, and body template.
Returns (effective_url, effective_headers, effective_body).
#_make_http_request
async def _make_http_request(*, method: str, url: str, headers: dict[str, str] | None, body: str | None, timeout_ceiling: int) -> tuple[httpx.Response, int]Execute the HTTP request. Returns (response, elapsed_ms).
#build_http_tool
def build_http_tool(definition: DataToolDefinition, deps: ToolDeps, timeout_ceiling: int=30) -> ToolBuild a Tool from a DataToolDefinition with HttpExecution config.
Args:
definition: A validated DataToolDefinition withtype = "http".deps: Session-scoped dependencies (secret_registry,
preview_threshold, preview_lines).
timeout_ceiling: Maximum HTTP timeout in seconds.
Returns:
- A Tool instance ready for execution pipeline wrapping.