Skip to content

Public API

This page is generated from the installed source. The guides explain when and why to use each part.

Execution

pilota.api.agent 

agent(task: str, *, model: Model, returns: type[T], context: Mapping[str, Any] | None = None, resources: Mapping[str, Resource] | None = None, instructions: Sequence[InstructionInput] | None = None, tools: Sequence[Tool] | None = None, workspace: WorkspaceHandle | None = None, limits: Limits | None = None, cancel_event: Event | None = None) -> T
Source code in src/pilota/api.py 
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
def agent(
    task: str,
    *,
    model: Model,
    returns: type[T],
    context: Mapping[str, Any] | None = None,
    resources: Mapping[str, Resource] | None = None,
    instructions: Sequence[InstructionInput] | None = None,
    tools: Sequence[Tool] | None = None,
    workspace: WorkspaceHandle | None = None,
    limits: Limits | None = None,
    cancel_event: Event | None = None,
) -> T:
    return run_agent(
        task,
        model=model,
        returns=returns,
        context=context,
        resources=resources,
        instructions=instructions,
        tools=tools,
        workspace=workspace,
        limits=limits,
        cancel_event=cancel_event,
    ).value

pilota.execution.run_agent 

run_agent(task: str, *, model: Model, returns: type[T], context: Mapping[str, Any] | None = None, resources: Mapping[str, Resource] | None = None, instructions: Sequence[InstructionInput] | None = None, tools: Sequence[Tool] | None = None, workspace: WorkspaceHandle | None = None, limits: Limits | None = None, cancel_event: Event | None = None) -> Execution[T]
Source code in src/pilota/execution.py 
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
def run_agent(
    task: str,
    *,
    model: Model,
    returns: type[T],
    context: Mapping[str, Any] | None = None,
    resources: Mapping[str, Resource] | None = None,
    instructions: Sequence[InstructionInput] | None = None,
    tools: Sequence[Tool] | None = None,
    workspace: WorkspaceHandle | None = None,
    limits: Limits | None = None,
    cancel_event: Event | None = None,
) -> Execution[T]:
    execution_id = str(uuid4())
    recorder = TraceRecorder()
    selected_limits = limits or Limits()
    deadline = (
        monotonic() + selected_limits.wall_time.total_seconds()
        if selected_limits.wall_time is not None
        else None
    )
    codec = ResultCodec(returns)
    rendered_instructions = render_instructions(
        tuple(instructions or ()),
        execution_id=execution_id,
    )
    selected_tools = (*tuple(tools or ()), *resource_tools(resources or {}))
    tool_registry = _build_tool_registry(selected_tools)
    tool_call_history: list[ToolCall] = []
    observations: list[Observation] = []
    model_calls = 0
    tool_calls = 0
    result_rejections = 0
    workspace_snapshot = workspace.snapshot() if workspace is not None else {}

    recorder.record("execution.started", {"task": task})

    def reject_result(attempt: FinalResult, error: ResultContractError) -> None:
        nonlocal result_rejections
        recorder.record("result.rejected", {"error": str(error)})
        if result_rejections >= selected_limits.result_retries:
            artifacts = _current_artifacts(workspace, workspace_snapshot)
            _record_artifacts(recorder, artifacts)
            recorder.record("execution.limit_exhausted", {"limit": "result_retries"})
            raise ResultContractError(
                str(error),
                execution_id=execution_id,
                artifacts=tuple(str(artifact.path) for artifact in artifacts),
                trace=recorder.snapshot(),
            ) from error
        result_rejections += 1
        call_id = attempt.id or str(uuid4())
        tool_call_history.append(
            ToolCall(call_id, RESULT_TOOL_NAME, attempt.value, attempt.provider_data)
        )
        observations.append(Observation(call_id, RESULT_TOOL_NAME, error=str(error)))

    while model_calls < selected_limits.model_calls:
        _check_execution_boundary(
            execution_id=execution_id,
            boundary="before_model",
            cancel_event=cancel_event,
            deadline=deadline,
            recorder=recorder,
            workspace=workspace,
            workspace_snapshot=workspace_snapshot,
        )
        request = ModelRequest(
            execution_id=execution_id,
            task=task,
            context=context or {},
            instructions=rendered_instructions,
            tools=tuple(tool.spec for tool in selected_tools),
            result_schema=codec.schema(),
            tool_calls=tuple(tool_call_history),
            observations=tuple(observations),
        )
        recorder.record("model.requested", {"call": model_calls + 1})
        try:
            action = model.respond(request)
        except AgentError as error:
            recorder.record("model.failed", {"error": str(error)})
            try:
                _check_execution_boundary(
                    execution_id=execution_id,
                    boundary="after_model",
                    cancel_event=cancel_event,
                    deadline=deadline,
                    recorder=recorder,
                    workspace=workspace,
                    workspace_snapshot=workspace_snapshot,
                )
            except AgentError as boundary_error:
                raise boundary_error from error
            artifacts = _current_artifacts(workspace, workspace_snapshot)
            _record_artifacts(recorder, artifacts)
            error.execution_id = execution_id
            error.artifacts = tuple(str(artifact.path) for artifact in artifacts)
            error.trace = recorder.snapshot()
            raise
        model_calls += 1
        _check_execution_boundary(
            execution_id=execution_id,
            boundary="after_model",
            cancel_event=cancel_event,
            deadline=deadline,
            recorder=recorder,
            workspace=workspace,
            workspace_snapshot=workspace_snapshot,
        )

        if isinstance(action, tuple) and (
            not action or not all(isinstance(call, ToolCall) for call in action)
        ):
            raise ConfigurationError(
                "Model returned an invalid tool-call batch", execution_id=execution_id
            )
        if isinstance(action, tuple) and len(action) == 1:
            action = action[0]
        if isinstance(action, ToolCall) and action.name == RESULT_TOOL_NAME:
            action = FinalResult(action.arguments, action.id, action.provider_data)

        if isinstance(action, FinalResult):
            try:
                value = codec.decode(action.value, execution_id=execution_id)
            except ResultContractError as error:
                reject_result(action, error)
                continue
            _check_execution_boundary(
                execution_id=execution_id,
                boundary="after_result_validation",
                cancel_event=cancel_event,
                deadline=deadline,
                recorder=recorder,
                workspace=workspace,
                workspace_snapshot=workspace_snapshot,
            )
            artifacts = _current_artifacts(workspace, workspace_snapshot)
            recorder.record("result.produced")
            _record_artifacts(recorder, artifacts)
            recorder.record("execution.completed")
            return Execution(
                id=execution_id,
                value=value,
                artifacts=artifacts,
                trace=recorder.snapshot(),
                usage=Usage(model_calls=model_calls, tool_calls=tool_calls),
                status=ExecutionStatus.SUCCEEDED,
            )

        if not isinstance(action, (ToolCall, tuple)):
            raise ConfigurationError(
                f"Model returned unsupported action: {type(action).__name__}",
                execution_id=execution_id,
            )

        batch = action if isinstance(action, tuple) else (action,)
        for action in (
            *[call for call in batch if call.name != RESULT_TOOL_NAME],
            *[call for call in batch if call.name == RESULT_TOOL_NAME],
        ):
            if action.name == RESULT_TOOL_NAME:
                reject_result(
                    FinalResult(action.arguments, action.id, action.provider_data),
                    ResultContractError(
                        "Call final_result alone, not together with other tool calls."
                    ),
                )
                continue
            tool_call_history.append(action)

            if tool_calls >= selected_limits.tool_calls:
                _raise_limit_error(
                    f"Tool-call limit of {selected_limits.tool_calls} was exhausted",
                    execution_id=execution_id,
                    limit="tool_calls",
                    recorder=recorder,
                    workspace=workspace,
                    workspace_snapshot=workspace_snapshot,
                )

            tool = tool_registry.get(action.name)
            if tool is None:
                message = f"Tool is not available: {action.name}"
                recorder.record("tool.denied", {"tool": action.name, "reason": message})
                observations.append(
                    Observation(
                        call_id=action.id,
                        tool_name=action.name,
                        error=message,
                    )
                )
                continue

            _check_execution_boundary(
                execution_id=execution_id,
                boundary="before_tool",
                cancel_event=cancel_event,
                deadline=deadline,
                recorder=recorder,
                workspace=workspace,
                workspace_snapshot=workspace_snapshot,
            )
            recorder.record("tool.started", {"tool": action.name, "call_id": action.id})
            tool_calls += 1

            try:
                output = tool.invoke(
                    action.arguments,
                    ToolContext(execution_id=execution_id, workspace=workspace),
                )
            except PermissionDenied as error:
                recorder.record(
                    "tool.denied",
                    {"tool": action.name, "call_id": action.id, "error": str(error)},
                )
                observations.append(
                    Observation(
                        call_id=action.id,
                        tool_name=action.name,
                        error=str(error),
                    )
                )
            except AgentError as error:
                recorder.record(
                    "tool.failed",
                    {"tool": action.name, "call_id": action.id, "error": str(error)},
                )
                observations.append(
                    Observation(
                        call_id=action.id,
                        tool_name=action.name,
                        error=str(error),
                    )
                )
            except Exception as error:
                wrapped = ToolError(
                    f"Tool {action.name!r} failed: {error}",
                    execution_id=execution_id,
                )
                recorder.record(
                    "tool.failed",
                    {"tool": action.name, "call_id": action.id, "error": str(wrapped)},
                )
                observations.append(
                    Observation(
                        call_id=action.id,
                        tool_name=action.name,
                        error=str(wrapped),
                    )
                )
            else:
                recorder.record("tool.completed", {"tool": action.name, "call_id": action.id})
                observations.append(
                    Observation(
                        call_id=action.id,
                        tool_name=action.name,
                        output=output,
                    )
                )
            _check_execution_boundary(
                execution_id=execution_id,
                boundary="after_tool",
                cancel_event=cancel_event,
                deadline=deadline,
                recorder=recorder,
                workspace=workspace,
                workspace_snapshot=workspace_snapshot,
            )

    _check_execution_boundary(
        execution_id=execution_id,
        boundary="after_loop",
        cancel_event=cancel_event,
        deadline=deadline,
        recorder=recorder,
        workspace=workspace,
        workspace_snapshot=workspace_snapshot,
    )
    _raise_limit_error(
        f"Model-call limit of {selected_limits.model_calls} was exhausted",
        execution_id=execution_id,
        limit="model_calls",
        recorder=recorder,
        workspace=workspace,
        workspace_snapshot=workspace_snapshot,
    )

Models and execution records

pilota.models.Limits dataclass 

Limits(model_calls: int = 30, tool_calls: int = 100, wall_time: timedelta | None = None, result_retries: int = 2)

Attributes

model_calls class-attribute instance-attribute 

model_calls: int = 30

tool_calls class-attribute instance-attribute 

tool_calls: int = 100

wall_time class-attribute instance-attribute 

wall_time: timedelta | None = None

result_retries class-attribute instance-attribute 

result_retries: int = 2

Methods:

__post_init__ 

__post_init__() -> None
Source code in src/pilota/models.py 
29
30
31
32
33
34
35
36
37
38
39
40
def __post_init__(self) -> None:
    if self.wall_time is not None:
        if not isinstance(self.wall_time, timedelta):
            raise TypeError("wall_time must be a datetime.timedelta")
        if self.wall_time <= timedelta(0):
            raise ValueError("wall_time must be greater than zero")
    if self.model_calls < 1:
        raise ValueError("model_calls must be at least 1")
    if self.tool_calls < 0:
        raise ValueError("tool_calls cannot be negative")
    if self.result_retries < 0:
        raise ValueError("result_retries cannot be negative")

pilota.models.Execution dataclass 

Execution(id: str, value: T, artifacts: tuple[Artifact, ...], trace: Trace, usage: Usage, status: ExecutionStatus = SUCCEEDED)

Attributes

id instance-attribute 

id: str

value instance-attribute 

value: T

artifacts instance-attribute 

artifacts: tuple[Artifact, ...]

trace instance-attribute 

trace: Trace

usage instance-attribute 

usage: Usage

status class-attribute instance-attribute 

status: ExecutionStatus = ExecutionStatus.SUCCEEDED

Methods:

pilota.models.ExecutionStatus

Attributes

SUCCEEDED class-attribute instance-attribute 

SUCCEEDED = 'succeeded'

FAILED class-attribute instance-attribute 

FAILED = 'failed'

CANCELLED class-attribute instance-attribute 

CANCELLED = 'cancelled'

pilota.models.Artifact dataclass 

Artifact(path: Path, operation: Literal['created', 'modified', 'deleted'])

Attributes

path instance-attribute 

path: Path

operation instance-attribute 

operation: Literal['created', 'modified', 'deleted']

Methods:

pilota.models.Usage dataclass 

Usage(model_calls: int = 0, tool_calls: int = 0)

Attributes

model_calls class-attribute instance-attribute 

model_calls: int = 0

tool_calls class-attribute instance-attribute 

tool_calls: int = 0

Methods:

pilota.models.Trace dataclass 

Trace(events: tuple[TraceEvent, ...])

Attributes

events instance-attribute 

events: tuple[TraceEvent, ...]

Methods:

pilota.models.TraceEvent dataclass 

TraceEvent(kind: str, data: Mapping[str, Any] = dict(), timestamp: datetime = (lambda: now(UTC))())

Attributes

kind instance-attribute 

kind: str

data class-attribute instance-attribute 

data: Mapping[str, Any] = field(default_factory=dict)

timestamp class-attribute instance-attribute 

timestamp: datetime = field(default_factory=lambda: datetime.now(UTC))

Methods:

pilota.models.ToolSpec dataclass 

ToolSpec(name: str, description: str, input_schema: Mapping[str, Any])

Attributes

name instance-attribute 

name: str

description instance-attribute 

description: str

input_schema instance-attribute 

input_schema: Mapping[str, Any]

Methods:

pilota.models.ModelRequest dataclass 

ModelRequest(execution_id: str, task: str, context: Mapping[str, Any], instructions: Sequence[str], tools: Sequence[ToolSpec], result_schema: Mapping[str, Any], tool_calls: Sequence[ToolCall], observations: Sequence[Observation])

Attributes

execution_id instance-attribute 

execution_id: str

task instance-attribute 

task: str

context instance-attribute 

context: Mapping[str, Any]

instructions instance-attribute 

instructions: Sequence[str]

tools instance-attribute 

tools: Sequence[ToolSpec]

result_schema instance-attribute 

result_schema: Mapping[str, Any]

tool_calls instance-attribute 

tool_calls: Sequence[ToolCall]

observations instance-attribute 

observations: Sequence[Observation]

Methods:

pilota.models.Observation dataclass 

Observation(call_id: str, tool_name: str, output: Any | None = None, error: str | None = None)

Attributes

call_id instance-attribute 

call_id: str

tool_name instance-attribute 

tool_name: str

output class-attribute instance-attribute 

output: Any | None = None

error class-attribute instance-attribute 

error: str | None = None

Methods:

pilota.models.ToolCall dataclass 

ToolCall(id: str, name: str, arguments: Any, provider_data: Mapping[str, Any] = dict())

Attributes

id instance-attribute 

id: str

name instance-attribute 

name: str

arguments instance-attribute 

arguments: Any

provider_data class-attribute instance-attribute 

provider_data: Mapping[str, Any] = field(default_factory=dict, repr=False, compare=False)

Methods:

pilota.models.FinalResult dataclass 

FinalResult(value: Any, id: str | None = None, provider_data: Mapping[str, Any] = dict())

Attributes

value instance-attribute 

value: Any

id class-attribute instance-attribute 

id: str | None = None

provider_data class-attribute instance-attribute 

provider_data: Mapping[str, Any] = field(default_factory=dict, repr=False, compare=False)

Methods:

Providers

pilota.providers.Model

Methods:

respond 

respond(request: ModelRequest) -> ModelAction
Source code in src/pilota/providers/__init__.py 
14
def respond(self, request: ModelRequest) -> ModelAction: ...

pilota.providers.ScriptedModel 

ScriptedModel(actions: Iterable[ModelAction])

Return a predetermined sequence of actions for deterministic tests.

Source code in src/pilota/providers/__init__.py 
20
21
22
def __init__(self, actions: Iterable[ModelAction]) -> None:
    self._actions = deque(actions)
    self.requests: list[ModelRequest] = []

Attributes

requests instance-attribute 

requests: list[ModelRequest] = []

Methods:

respond 

respond(request: ModelRequest) -> ModelAction
Source code in src/pilota/providers/__init__.py 
24
25
26
27
28
def respond(self, request: ModelRequest) -> ModelAction:
    self.requests.append(request)
    if not self._actions:
        raise ProviderError("ScriptedModel has no actions remaining")
    return self._actions.popleft()

pilota.providers.openrouter.OpenRouterModel 

OpenRouterModel(model: str, *, api_key: str | None = None, base_url: str = DEFAULT_BASE_URL, timeout: float = 60.0, http_referer: str | None = None, app_title: str | None = 'Pilota', client: Client | None = None)

Execute Pilota model requests through OpenRouter's Chat Completions API.

Source code in src/pilota/providers/openrouter.py 
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
def __init__(
    self,
    model: str,
    *,
    api_key: str | None = None,
    base_url: str = DEFAULT_BASE_URL,
    timeout: float = 60.0,
    http_referer: str | None = None,
    app_title: str | None = "Pilota",
    client: httpx.Client | None = None,
) -> None:
    selected_api_key = api_key or os.environ.get("OPENROUTER_API_KEY")
    if not selected_api_key:
        raise ConfigurationError(
            "OpenRouter requires api_key or the OPENROUTER_API_KEY environment variable"
        )
    if not model.strip():
        raise ConfigurationError("OpenRouter model must not be empty")

    self.model = model
    self._api_key = selected_api_key
    self._base_url = base_url.rstrip("/")
    self._http_referer = http_referer
    self._app_title = app_title
    self._client = client or httpx.Client(timeout=timeout)
    self._owns_client = client is None

Attributes

model instance-attribute 

model = model

Methods:

respond 

respond(request: ModelRequest) -> ModelAction
Source code in src/pilota/providers/openrouter.py 
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
def respond(self, request: ModelRequest) -> ModelAction:
    try:
        response = self._client.post(
            f"{self._base_url}/chat/completions",
            headers=self._headers(),
            json=self._payload(request),
        )
    except httpx.HTTPError as error:
        raise ProviderError(
            f"OpenRouter request failed: {error}",
            execution_id=request.execution_id,
        ) from error

    if not response.is_success:
        raise ProviderError(
            _error_message(response),
            execution_id=request.execution_id,
        )

    try:
        body = response.json()
    except ValueError as error:
        raise ProviderError(
            "OpenRouter returned a non-JSON response",
            execution_id=request.execution_id,
        ) from error

    return _parse_action(
        body,
        execution_id=request.execution_id,
        wrapped_result=request.result_schema.get("type") != "object",
    )

close 

close() -> None
Source code in src/pilota/providers/openrouter.py 
91
92
93
def close(self) -> None:
    if self._owns_client:
        self._client.close()

__enter__ 

__enter__() -> OpenRouterModel
Source code in src/pilota/providers/openrouter.py 
95
96
def __enter__(self) -> OpenRouterModel:
    return self

__exit__ 

__exit__(*args: object) -> None
Source code in src/pilota/providers/openrouter.py 
98
99
def __exit__(self, *args: object) -> None:
    self.close()

Instructions

pilota.instructions.Instruction

Methods:

render 

render() -> str
Source code in src/pilota/instructions.py 
14
def render(self) -> str: ...

pilota.instructions.InstructionFile dataclass 

InstructionFile(path: str | Path, encoding: str = 'utf-8')

Attributes

path instance-attribute 

path: str | Path

encoding class-attribute instance-attribute 

encoding: str = 'utf-8'

Methods:

render 

render() -> str
Source code in src/pilota/instructions.py 
25
26
27
28
29
30
31
def render(self) -> str:
    try:
        return Path(self.path).read_text(encoding=self.encoding)
    except (LookupError, OSError, UnicodeError) as error:
        raise ConfigurationError(
            f"Could not load instruction file {self.path!s}: {error}"
        ) from error

Resources

pilota.resources.Resource

Attributes

operations property 

operations: Mapping[str, ResourceOperation]

Methods:

invoke 

invoke(operation: str, arguments: Mapping[str, Any]) -> Any
Source code in src/pilota/resources/base.py 
25
def invoke(self, operation: str, arguments: Mapping[str, Any]) -> Any: ...

pilota.resources.ResourceOperation dataclass 

ResourceOperation(name: str, description: str, input_schema: Mapping[str, Any])

Attributes

name instance-attribute 

name: str

description instance-attribute 

description: str

input_schema instance-attribute 

input_schema: Mapping[str, Any]

Methods:

pilota.resources.DirectoryResource 

DirectoryResource(root: str | Path, *, allow: Collection[DirectoryOperation] = ('list', 'read'), deny_globs: Collection[str] = ())
Source code in src/pilota/resources/directory.py 
17
18
19
20
21
22
23
24
25
26
def __init__(
    self,
    root: str | Path,
    *,
    allow: Collection[DirectoryOperation] = ("list", "read"),
    deny_globs: Collection[str] = (),
) -> None:
    self.root = Path(root).resolve()
    self._allow = frozenset(allow)
    self._deny_globs = tuple(deny_globs)

Attributes

root instance-attribute 

root = Path(root).resolve()

operations property 

operations: Mapping[str, ResourceOperation]

Methods:

invoke 

invoke(operation: str, arguments: Mapping[str, Any]) -> Any
Source code in src/pilota/resources/directory.py 
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
def invoke(self, operation: str, arguments: Mapping[str, Any]) -> Any:
    if operation not in self._allow:
        raise PermissionDenied(f"Directory operation is not allowed: {operation}")
    path = arguments.get("path", ".")
    if not isinstance(path, str):
        raise ToolError("Directory path must be a string")
    target = self._resolve(path)
    if operation == "list":
        if not target.exists():
            return []
        if target.is_file():
            return [str(target.relative_to(self.root))]
        return sorted(
            str(candidate.relative_to(self.root))
            for candidate in target.rglob("*")
            if candidate.is_file()
        )
    if operation == "read":
        return target.read_text()
    raise ToolError(f"Unsupported directory operation: {operation}")

Tools

pilota.tools.Tool

Attributes

spec property 

spec: ToolSpec

Methods:

invoke 

invoke(arguments: Mapping[str, Any], context: ToolContext) -> Any
Source code in src/pilota/tools/base.py 
40
def invoke(self, arguments: Mapping[str, Any], context: ToolContext) -> Any: ...

pilota.tools.ToolContext dataclass 

ToolContext(execution_id: str, workspace: WorkspaceHandle | None)

Attributes

execution_id instance-attribute 

execution_id: str

workspace instance-attribute 

workspace: WorkspaceHandle | None

Methods:

pilota.tools.WorkspaceHandle

Methods:

read_text 

read_text(path: str) -> str
Source code in src/pilota/tools/base.py 
14
def read_text(self, path: str) -> str: ...

write_text 

write_text(path: str, content: str) -> None
Source code in src/pilota/tools/base.py 
16
def write_text(self, path: str, content: str) -> None: ...

list_files 

list_files(path: str = '.') -> list[str]
Source code in src/pilota/tools/base.py 
18
def list_files(self, path: str = ".") -> list[str]: ...

snapshot 

snapshot() -> Mapping[str, str]
Source code in src/pilota/tools/base.py 
20
def snapshot(self) -> Mapping[str, str]: ...

artifacts_since 

artifacts_since(snapshot: Mapping[str, str]) -> list[Any]
Source code in src/pilota/tools/base.py 
22
def artifacts_since(self, snapshot: Mapping[str, str]) -> list[Any]: ...

pilota.tools.ProcessWorkspaceHandle

Methods:

execution_directory 

execution_directory(path: str = '.') -> Path
Source code in src/pilota/tools/base.py 
27
def execution_directory(self, path: str = ".") -> Path: ...

pilota.tools.Editor

Attributes

spec property 

spec: ToolSpec

Methods:

invoke 

invoke(arguments: Mapping[str, Any], context: ToolContext) -> Any
Source code in src/pilota/tools/editor.py 
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
def invoke(self, arguments: Mapping[str, Any], context: ToolContext) -> Any:
    if context.workspace is None:
        raise ConfigurationError("Editor requires a workspace")

    operation = arguments.get("operation")
    path = arguments.get("path", ".")
    if not isinstance(path, str):
        raise ToolError("Editor path must be a string")

    if operation == "list":
        return context.workspace.list_files(path)
    if operation == "read":
        return context.workspace.read_text(path)
    if operation == "write":
        content = arguments.get("content")
        if not isinstance(content, str):
            raise ToolError("Editor write requires string content")
        context.workspace.write_text(path, content)
        return {"path": path, "written": len(content)}
    raise ToolError(f"Unsupported editor operation: {operation!r}")

pilota.tools.Shell 

Shell(commands: Mapping[str, Sequence[str]], *, timeout: float = 30.0, max_output_chars: int = 20000, environment: Mapping[str, str] | None = None)

Execute explicitly configured command aliases inside a workspace.

Source code in src/pilota/tools/shell.py 
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
def __init__(
    self,
    commands: Mapping[str, Sequence[str]],
    *,
    timeout: float = 30.0,
    max_output_chars: int = 20_000,
    environment: Mapping[str, str] | None = None,
) -> None:
    if timeout <= 0:
        raise ValueError("timeout must be greater than zero")
    if max_output_chars < 1:
        raise ValueError("max_output_chars must be at least one")

    self._commands = _validate_commands(commands)
    self._timeout = timeout
    self._max_output_chars = max_output_chars
    self._environment = {
        "PATH": os.defpath,
        **_validate_environment(environment or {}),
    }

Attributes

spec property 

spec: ToolSpec

Methods:

invoke 

invoke(arguments: Mapping[str, Any], context: ToolContext) -> Any
Source code in src/pilota/tools/shell.py 
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
def invoke(self, arguments: Mapping[str, Any], context: ToolContext) -> Any:
    if context.workspace is None:
        raise ConfigurationError("Shell requires a workspace")
    if not isinstance(context.workspace, ProcessWorkspaceHandle):
        raise ConfigurationError(
            "Shell requires a process-backed workspace",
            execution_id=context.execution_id,
        )

    alias = arguments.get("command")
    if not isinstance(alias, str):
        raise ToolError("Shell command must be a string")
    command = self._commands.get(alias)
    if command is None:
        raise ToolError(f"Shell command is not allowed: {alias}")

    raw_args = arguments.get("args", [])
    if not isinstance(raw_args, list) or not all(isinstance(item, str) for item in raw_args):
        raise ToolError("Shell args must be a list of strings")
    if any("\0" in item for item in raw_args):
        raise ToolError("Shell args must not contain null bytes")

    cwd = arguments.get("cwd", ".")
    if not isinstance(cwd, str):
        raise ToolError("Shell cwd must be a string")

    try:
        execution_directory = context.workspace.execution_directory(cwd)
    except (OSError, ValueError) as error:
        raise ToolError(str(error), execution_id=context.execution_id) from error

    try:
        completed = subprocess.run(
            (*command, *raw_args),
            cwd=execution_directory,
            env=self._environment,
            stdin=subprocess.DEVNULL,
            capture_output=True,
            check=False,
            timeout=self._timeout,
        )
    except subprocess.TimeoutExpired as error:
        raise ToolTimeoutError(
            f"Shell command {alias!r} exceeded its {self._timeout:g}s timeout",
            execution_id=context.execution_id,
        ) from error
    except OSError as error:
        raise ToolError(
            f"Shell command {alias!r} could not start: {error}",
            execution_id=context.execution_id,
        ) from error

    stdout, stdout_truncated = _decode_output(
        completed.stdout,
        max_chars=self._max_output_chars,
    )
    stderr, stderr_truncated = _decode_output(
        completed.stderr,
        max_chars=self._max_output_chars,
    )
    return {
        "command": alias,
        "exit_code": completed.returncode,
        "stdout": stdout,
        "stderr": stderr,
        "stdout_truncated": stdout_truncated,
        "stderr_truncated": stderr_truncated,
    }

Workspaces

pilota.workspaces.InMemoryWorkspace 

InMemoryWorkspace(files: Mapping[str, str] | None = None, *, allow: Collection[Operation] = ('list', 'read', 'create', 'edit'), deny_globs: Collection[str] = ())
Source code in src/pilota/workspaces/memory.py 
13
14
15
16
17
18
19
20
21
22
23
def __init__(
    self,
    files: Mapping[str, str] | None = None,
    *,
    allow: Collection[Operation] = ("list", "read", "create", "edit"),
    deny_globs: Collection[str] = (),
) -> None:
    self._files = {
        str(normalize_relative_path(path)): content for path, content in (files or {}).items()
    }
    self._policy = WorkspacePolicy(allow=allow, deny_globs=deny_globs)

Methods:

list_files 

list_files(path: str = '.') -> list[str]
Source code in src/pilota/workspaces/memory.py 
25
26
27
28
29
30
31
32
def list_files(self, path: str = ".") -> list[str]:
    self._policy.authorize("list", path)
    prefix = normalize_relative_path(path)
    if str(prefix) == ".":
        return sorted(self._files)
    return sorted(
        name for name in self._files if normalize_relative_path(name).is_relative_to(prefix)
    )

read_text 

read_text(path: str) -> str
Source code in src/pilota/workspaces/memory.py 
34
35
36
37
def read_text(self, path: str) -> str:
    self._policy.authorize("read", path)
    normalized = str(normalize_relative_path(path))
    return self._files[normalized]

write_text 

write_text(path: str, content: str) -> None
Source code in src/pilota/workspaces/memory.py 
39
40
41
42
43
def write_text(self, path: str, content: str) -> None:
    normalized = str(normalize_relative_path(path))
    operation: Operation = "edit" if normalized in self._files else "create"
    self._policy.authorize(operation, path)
    self._files[normalized] = content

snapshot 

snapshot() -> Mapping[str, str]
Source code in src/pilota/workspaces/memory.py 
45
46
def snapshot(self) -> Mapping[str, str]:
    return dict(self._files)

artifacts_since 

artifacts_since(snapshot: Mapping[str, str]) -> list[Artifact]
Source code in src/pilota/workspaces/memory.py 
48
49
50
51
52
53
54
55
56
57
58
def artifacts_since(self, snapshot: Mapping[str, str]) -> list[Artifact]:
    artifacts: list[Artifact] = []
    before = dict(snapshot)
    for path, content in self._files.items():
        if path not in before:
            artifacts.append(Artifact(path=Path(path), operation="created"))
        elif before[path] != content:
            artifacts.append(Artifact(path=Path(path), operation="modified"))
    for path in before.keys() - self._files.keys():
        artifacts.append(Artifact(path=Path(path), operation="deleted"))
    return sorted(artifacts, key=lambda artifact: str(artifact.path))

pilota.workspaces.DirectoryWorkspace 

DirectoryWorkspace(root: str | Path, *, allow: Collection[Operation], deny_globs: Collection[str] = ())
Source code in src/pilota/workspaces/directory.py 
14
15
16
17
18
19
20
21
22
def __init__(
    self,
    root: str | Path,
    *,
    allow: Collection[Operation],
    deny_globs: Collection[str] = (),
) -> None:
    self.root = Path(root).resolve()
    self._policy = WorkspacePolicy(allow=allow, deny_globs=deny_globs)

Attributes

root instance-attribute 

root = Path(root).resolve()

Methods:

list_files 

list_files(path: str = '.') -> list[str]
Source code in src/pilota/workspaces/directory.py 
24
25
26
27
28
29
30
31
32
33
34
35
def list_files(self, path: str = ".") -> list[str]:
    self._policy.authorize("list", path)
    target = self._resolve(path)
    if not target.exists():
        return []
    if target.is_file():
        return [str(target.relative_to(self.root))]
    return sorted(
        str(candidate.relative_to(self.root))
        for candidate in target.rglob("*")
        if candidate.is_file()
    )

read_text 

read_text(path: str) -> str
Source code in src/pilota/workspaces/directory.py 
37
38
39
def read_text(self, path: str) -> str:
    self._policy.authorize("read", path)
    return self._resolve(path).read_text()

write_text 

write_text(path: str, content: str) -> None
Source code in src/pilota/workspaces/directory.py 
41
42
43
44
45
46
def write_text(self, path: str, content: str) -> None:
    target = self._resolve(path)
    operation: Operation = "edit" if target.exists() else "create"
    self._policy.authorize(operation, path)
    target.parent.mkdir(parents=True, exist_ok=True)
    target.write_text(content)

execution_directory 

execution_directory(path: str = '.') -> Path
Source code in src/pilota/workspaces/directory.py 
48
49
50
51
52
53
54
55
def execution_directory(self, path: str = ".") -> Path:
    self._policy.authorize("execute", path)
    target = self._resolve(path)
    if not target.exists():
        raise ValueError(f"Workspace execution directory does not exist: {path}")
    if not target.is_dir():
        raise ValueError(f"Workspace execution path is not a directory: {path}")
    return target

snapshot 

snapshot() -> Mapping[str, str]
Source code in src/pilota/workspaces/directory.py 
57
58
59
60
61
62
63
64
def snapshot(self) -> Mapping[str, str]:
    if not self.root.exists():
        return {}
    return {
        str(path.relative_to(self.root)): sha256(path.read_bytes()).hexdigest()
        for path in self.root.rglob("*")
        if path.is_file()
    }

artifacts_since 

artifacts_since(snapshot: Mapping[str, str]) -> list[Artifact]
Source code in src/pilota/workspaces/directory.py 
66
67
68
69
70
71
72
73
74
75
76
def artifacts_since(self, snapshot: Mapping[str, str]) -> list[Artifact]:
    current = self.snapshot()
    artifacts: list[Artifact] = []
    for path, digest in current.items():
        if path not in snapshot:
            artifacts.append(Artifact(path=Path(path), operation="created"))
        elif snapshot[path] != digest:
            artifacts.append(Artifact(path=Path(path), operation="modified"))
    for path in snapshot.keys() - current.keys():
        artifacts.append(Artifact(path=Path(path), operation="deleted"))
    return sorted(artifacts, key=lambda artifact: str(artifact.path))

pilota.workspaces.WorkspacePolicy 

WorkspacePolicy(*, allow: Collection[Operation], deny_globs: Collection[str] = ())
Source code in src/pilota/workspaces/base.py 
15
16
17
18
19
20
21
22
def __init__(
    self,
    *,
    allow: Collection[Operation],
    deny_globs: Collection[str] = (),
) -> None:
    self.allow = frozenset(allow)
    self.deny_globs = tuple(deny_globs)

Attributes

allow instance-attribute 

allow = frozenset(allow)

deny_globs instance-attribute 

deny_globs = tuple(deny_globs)

Methods:

authorize 

authorize(operation: Operation, path: str) -> None
Source code in src/pilota/workspaces/base.py 
24
25
26
27
28
29
def authorize(self, operation: Operation, path: str) -> None:
    normalized = normalize_relative_path(path)
    if operation not in self.allow:
        raise PermissionDenied(f"Workspace operation is not allowed: {operation}")
    if any(normalized.match(pattern) for pattern in self.deny_globs):
        raise PermissionDenied(f"Workspace path is denied: {normalized}")