data_juicer_agents.core.tool.contracts module#

Runtime-agnostic tool contracts.

class data_juicer_agents.core.tool.contracts.ToolArtifact(path: str, description: str = '', kind: str = 'file', label: str = '')[source]#

Bases: object

Named artifact produced by a tool.

path: str#
description: str = ''#
kind: str = 'file'#
label: str = ''#
to_dict() → Dict[str, Any][source]#
__init__(path: str, description: str = '', kind: str = 'file', label: str = '') → None#
class data_juicer_agents.core.tool.contracts.ToolContext(working_dir: str = './.djx', env: Dict[str, str]=<factory>, artifacts_dir: str | None = None, runtime_values: Dict[str, ~typing.Any]=<factory>)[source]#

Bases: object

Execution context shared by all tool runtimes.

working_dir: str = './.djx'#
env: Dict[str, str]#
artifacts_dir: str | None = None#
runtime_values: Dict[str, Any]#
resolve_artifacts_dir() → Path[source]#
__init__(working_dir: str = './.djx', env: Dict[str, str]=<factory>, artifacts_dir: str | None = None, runtime_values: Dict[str, ~typing.Any]=<factory>) → None#
class data_juicer_agents.core.tool.contracts.ToolResult(ok: bool, summary: str = '', data: Dict[str, ~typing.Any]=<factory>, artifacts: List[ToolArtifact] = <factory>, error_type: str = '', error_message: str = '', next_actions: List[str] = <factory>)[source]#

Bases: object

Normalized tool execution result.

ok: bool#
summary: str = ''#
data: Dict[str, Any]#
artifacts: List[ToolArtifact]#
error_type: str = ''#
error_message: str = ''#
next_actions: List[str]#
classmethod success(*, summary: str = '', data: Dict[str, Any] | None = None, artifacts: Iterable[ToolArtifact] | None = None) → ToolResult[source]#
classmethod failure(*, summary: str, error_type: str, error_message: str = '', data: Dict[str, Any] | None = None, next_actions: Iterable[str] | None = None) → ToolResult[source]#
to_payload(*, action: str | None = None) → Dict[str, Any][source]#
__init__(ok: bool, summary: str = '', data: Dict[str, ~typing.Any]=<factory>, artifacts: List[ToolArtifact] = <factory>, error_type: str = '', error_message: str = '', next_actions: List[str] = <factory>) → None#
class data_juicer_agents.core.tool.contracts.ToolSpec(name: str, description: str, input_model: Type[BaseModel], output_model: Type[BaseModel] | None, executor: Callable[[ToolContext, BaseModel], ToolResult], tags: Tuple[str, ...] = (), effects: Literal['read', 'write', 'execute', 'external'] = 'read', confirmation: Literal['none', 'recommended', 'required'] = 'none')[source]#

Bases: object

Definition of one atomic tool.

name: str#
description: str#
input_model: Type[BaseModel]#
output_model: Type[BaseModel] | None#
executor: Callable[[ToolContext, BaseModel], ToolResult]#
tags: Tuple[str, ...] = ()#
effects: Literal['read', 'write', 'execute', 'external'] = 'read'#
confirmation: Literal['none', 'recommended', 'required'] = 'none'#
execute(ctx: ToolContext, raw_input: BaseModel | Dict[str, Any]) → ToolResult[source]#
__init__(name: str, description: str, input_model: Type[BaseModel], output_model: Type[BaseModel] | None, executor: Callable[[ToolContext, BaseModel], ToolResult], tags: Tuple[str, ...] = (), effects: Literal['read', 'write', 'execute', 'external'] = 'read', confirmation: Literal['none', 'recommended', 'required'] = 'none') → None#