artifactr.agent¶

artifactr's integration with pydantic-ai.

Give an agent the ArtifactWorkspace capability and deps_type=Session[...], then run it in threads with a Runner::

agent = Agent(
    "anthropic:claude-sonnet-5-5",
    deps_type=Session[None],
    capabilities=[ArtifactWorkspace(types=[Doc, Plan])],
)
runner = Runner(agent, app=None)
await runner.send(workspace, thread_id, "Draft a launch plan")

The capability¶

See The agent.

ArtifactWorkspace `dataclass` ¶

ArtifactWorkspace(
    types: Sequence[type[Artifact]],
    *,
    ask: bool = False,
    max_render_chars: int = 4000,
    max_summary_chars: int = 200,
)

Bases: AbstractCapability[Session[Any]]

Make an agent a participant in a workspace.

Parameters:

Name	Type	Description	Default
`types`	`Sequence[type[Artifact]]`	The artifact types the agent may create.	required
`ask`	`bool`	Include the `ask_user` tool, which pauses the run until someone answers. The agent's `output_type` must then include `DeferredToolRequests`.	`False`
`max_render_chars`	`int`	How much of each followed artifact's rendering to include in the instructions.	`4000`
`max_summary_chars`	`int`	How much of each tool call's arguments and result to record.	`200`

get_toolset ¶

get_toolset() -> FunctionToolset[Session[Any]]

Return the generic artifact tools.

get_instructions ¶

get_instructions() -> Any

Return the instructions, rendered fresh for every model request.

wrap_run `async` ¶

wrap_run(
    ctx: Context, *, handler: WrapRunHandler
) -> AgentRunResult[Any]

Record the run, brief the agent, and watch the workspace while it runs.

before_tool_execute `async` ¶

before_tool_execute(
    ctx: Context,
    *,
    call: ToolCallPart,
    tool_def: ToolDefinition,
    args: ValidatedToolArgs,
) -> ValidatedToolArgs

Record the tool call.

wrap_tool_execute `async` ¶

wrap_tool_execute(
    ctx: Context,
    *,
    call: ToolCallPart,
    tool_def: ToolDefinition,
    args: ValidatedToolArgs,
    handler: WrapToolExecuteHandler,
) -> Any

Record a tool's own ModelRetry or ToolFailed, which skip the error hook.

after_tool_execute `async` ¶

after_tool_execute(
    ctx: Context,
    *,
    call: ToolCallPart,
    tool_def: ToolDefinition,
    args: ValidatedToolArgs,
    result: Any,
) -> Any

Record the tool's result.

on_tool_execute_error `async` ¶

on_tool_execute_error(
    ctx: Context,
    *,
    call: ToolCallPart,
    tool_def: ToolDefinition,
    args: ValidatedToolArgs,
    error: Exception,
) -> Any

Turn rejections into retries the model can act on; record every failure.

Session `dataclass` ¶

Session(
    workspace: Workspace,
    thread_id: ThreadId,
    run_id: RunId,
    app: AppDepsT,
    trigger: Trigger = "api",
    watch_after: int | None = None,
)

The pydantic-ai deps of one agent run.

Tools receive it as ctx.deps. Its workspace handle acts as the thread's agent, so everything a tool commits is attributed to the agent and this run.

Create it with start rather than directly.

app `instance-attribute` ¶

app: AppDepsT

The application's own dependencies.

watch_after `class-attribute` `instance-attribute` ¶

watch_after: int | None = None

Watch the log for others' changes after this seq; None means from the run's start.

start `classmethod` ¶

start(
    workspace: Workspace,
    thread_id: ThreadId,
    *,
    app: AppDepsT,
    run_id: RunId | None = None,
    agent_name: str = "assistant",
    trigger: Trigger = "api",
    watch_after: int | None = None,
) -> Session[AppDepsT]

Return a session for a new (or resuming) run of the thread's agent.

Parameters:

Name	Type	Description	Default
`workspace`	`Workspace`	A handle on the workspace; the session derives the agent's handle.	required
`thread_id`	`ThreadId`	The thread the agent runs in.	required
`app`	`AppDepsT`	The application's own dependencies, available to tools as `ctx.deps.app`.	required
`run_id`	`RunId \| None`	The run's id; a new one is generated when omitted.	`None`
`agent_name`	`str`	How the agent is named in change notes and messages.	`'assistant'`
`trigger`	`Trigger`	What started the run.	`'api'`
`watch_after`	`int \| None`	The `seq` after which others' changes are delivered into the run.	`None`

Trigger `module-attribute` ¶

Trigger = Literal['message', 'resume', 'api']

What started a run: a posted message, answers to a pause, or a direct call.

load_history `async` ¶

load_history(
    workspace: Workspace, thread_id: ThreadId
) -> list[ModelMessage]

Return a thread's model history, to pass as message_history.

last_seen `async` ¶

last_seen(workspace: Workspace, thread_id: ThreadId) -> int

Return the seq up to which the thread's agent has been told what happened.

Running agents in threads¶

See Running the agent.

Runner ¶

Runner(
    agent: Agent[Session[AppDepsT], Any],
    *,
    app: AppDepsT,
    live: FanoutChannel | None = None,
    agent_name: str = "assistant",
    claim_ttl: timedelta = timedelta(seconds=30),
)

Runs an agent in workspace threads.

Parameters:

Name	Type	Description	Default
`agent`	`Agent[Session[AppDepsT], Any]`	The agent, whose `deps_type` is `Session[AppDepsT]` and which has the `ArtifactWorkspace` capability.	required
`app`	`AppDepsT`	The application's dependencies, passed to every run as `ctx.deps.app`.	required
`live`	`FanoutChannel \| None`	Where runs' live frames go. Defaults to an in-process fan-out.	`None`
`agent_name`	`str`	How the agent is named in the workspace.	`'assistant'`
`claim_ttl`	`timedelta`	How long a thread claim lasts without renewal, should this process die.	`timedelta(seconds=30)`

execute `async` ¶

execute(
    workspace: Workspace, command: Command | StopRun
) -> Outcome

Carry out any command the way every surface should.

Messages and answers go through send and answer, so they start, steer and resume runs; stop_run stops a run of this workspace; everything else is committed as-is.

Raises:

Type	Description
`Rejection`	If the command is rejected, or the run to stop is not in this workspace or not running in this process.

send `async` ¶

send(
    workspace: Workspace,
    thread_id: ThreadId,
    content: str,
    *,
    message_id: MessageId | None = None,
) -> Sent

Post a message as the workspace handle's actor, and act on it.

answer `async` ¶

answer(
    workspace: Workspace, command: AnswerDeferred
) -> Sent

Answer one of a paused run's requests, resuming the run once all are answered.

resume `async` ¶

resume(
    workspace: Workspace, run_id: RunId
) -> RunHandle | None

Resume a paused run whose requests are all answered; otherwise do nothing.

stop `async` ¶

stop(run_id: RunId) -> bool

Cancel a run of this process. Returns whether there was one to stop.

watch ¶

watch(run_id: RunId) -> AsyncIterator[LiveFrame]

Yield a run's live frames from now until it ends.

running ¶

running(thread_id: ThreadId) -> RunHandle | None

Return this process's run in a thread, if there is one.

RunHandle `dataclass` ¶

RunHandle(
    run_id: RunId,
    thread_id: ThreadId,
    task: Task[AgentRunResult[Any]],
)

A run started by a Runner.

wait `async` ¶

wait() -> AgentRunResult[Any]

Wait for the run to finish (or pause) and return its result.

Sent `dataclass` ¶

Sent(outcome: Recorded, run: RunHandle | None)

What posting a message, or an answer, did.

outcome `instance-attribute` ¶

outcome: Recorded

The recorded message or answer.

run `instance-attribute` ¶

run: RunHandle | None

The run it started or resumed; None if it steered a running run or awaits more answers.

Live output¶

See Live output.

ArtifactDraft `dataclass` ¶

ArtifactDraft(
    *,
    kind: str,
    snapshot: dict[str, Any],
    artifact_id: ArtifactId | None = None,
)

Bases: CustomEvent

A snapshot of an artifact that a tool is still generating.

Emit it from an application tool with await ctx.emit(ArtifactDraft(...)); it reaches live channels as a draft frame and is never stored.

forward_live ¶

forward_live(channel: LiveChannel) -> EventStreamHandler

Return a pydantic-ai event_stream_handler that sends live frames to channel.

The run id comes from the run's Session.

to_live ¶

to_live(event: AgentStreamEvent) -> list[LiveEvent]

Translate one pydantic-ai stream event into live events (often none).

LiveChannel ¶

Bases: Protocol

Where a run's live frames go.

send `async` ¶

send(frame: LiveFrame) -> None

Deliver one frame. Delivery is best-effort.

FanoutChannel ¶

FanoutChannel(
    *, buffer: int = 1024, remember_closed: int = 10000
)

An in-process channel that fans each run's frames out to its watchers.

It keeps the frames of runs in progress, so a watcher that attaches mid-run first receives what the run has produced so far. Watchers that fall behind lose their oldest frames rather than slowing the run.

Parameters:

Name	Type	Description	Default
`buffer`	`int`	How many frames each run keeps, and each watcher may fall behind.	`1024`
`remember_closed`	`int`	How many ended runs to remember, so late watchers end at once.	`10000`

send `async` ¶

send(frame: LiveFrame) -> None

Keep a frame for the run, and deliver it to the run's current watchers.

watch `async` ¶

watch(run_id: RunId) -> AsyncIterator[LiveFrame]

Yield the run's frames so far, then new ones until close is called for it.

Watching a run that has already ended stops at once.

close ¶

close(run_id: RunId) -> None

End every watcher of a run, and any that start watching it later.

NullChannel ¶

A channel that drops every frame, for headless runs.

send `async` ¶

send(frame: LiveFrame) -> None

Drop the frame.

Tool helpers¶

The generic tools, and helpers for writing your own.

artifact_tools ¶

artifact_tools(
    types: Sequence[type[Artifact]], *, ask: bool = False
) -> FunctionToolset[Session[Any]]

Build the generic toolset for the given artifact types.

Parameters:

Name	Type	Description	Default
`types`	`Sequence[type[Artifact]]`	The artifact types the agent may create; their JSON Schemas are described to the model once, so tool definitions never change between requests.	required
`ask`	`bool`	Whether to include `ask_user`, which pauses the run until someone answers. The agent's `output_type` must then include `DeferredToolRequests`.	`False`

describe_outcome ¶

describe_outcome(outcome: Applied | Proposed) -> str

Tell a model what its change did.

submit `async` ¶

submit(
    workspace: Workspace,
    change: EditArtifact | ArchiveArtifact | CreateArtifact,
    *,
    propose: bool = False,
    rationale: str | None = None,
) -> Applied | Proposed

Commit a change, or propose it for review when propose is set.

artifact_text ¶

artifact_text(artifact: Versioned[Artifact]) -> str

Render an artifact for a model: a header line, then its render_for_agent text.

list_artifacts_text `async` ¶

list_artifacts_text(
    workspace: Workspace, kind: str | None = None
) -> str

List a workspace's artifacts as text, one per line.

artifactr.agent¶

The capability¶

ArtifactWorkspace dataclass ¶

get_toolset ¶

get_instructions ¶

wrap_run async ¶

before_tool_execute async ¶

wrap_tool_execute async ¶

after_tool_execute async ¶

on_tool_execute_error async ¶

Session dataclass ¶

app instance-attribute ¶

watch_after class-attribute instance-attribute ¶

start classmethod ¶

Trigger module-attribute ¶

load_history async ¶

last_seen async ¶

Running agents in threads¶

Runner ¶

execute async ¶

send async ¶

answer async ¶

resume async ¶

stop async ¶

watch ¶

running ¶

RunHandle dataclass ¶

wait async ¶

Sent dataclass ¶

outcome instance-attribute ¶

run instance-attribute ¶

Live output¶

ArtifactDraft dataclass ¶

forward_live ¶

to_live ¶

LiveChannel ¶

send async ¶

FanoutChannel ¶

send async ¶

watch async ¶

close ¶

NullChannel ¶

send async ¶

Tool helpers¶

artifact_tools ¶

describe_outcome ¶

submit async ¶

artifact_text ¶

list_artifacts_text async ¶

ArtifactWorkspace `dataclass` ¶

wrap_run `async` ¶

before_tool_execute `async` ¶

wrap_tool_execute `async` ¶

after_tool_execute `async` ¶

on_tool_execute_error `async` ¶

Session `dataclass` ¶

app `instance-attribute` ¶

watch_after `class-attribute` `instance-attribute` ¶

start `classmethod` ¶

Trigger `module-attribute` ¶

load_history `async` ¶

last_seen `async` ¶

execute `async` ¶

send `async` ¶

answer `async` ¶

resume `async` ¶

stop `async` ¶

RunHandle `dataclass` ¶

wait `async` ¶

Sent `dataclass` ¶

outcome `instance-attribute` ¶

run `instance-attribute` ¶

ArtifactDraft `dataclass` ¶

send `async` ¶

send `async` ¶

watch `async` ¶

send `async` ¶

submit `async` ¶

list_artifacts_text `async` ¶