Runtime Capability Creation
Runtime capability creation lets an agent write, validate, and persist Pydantic AI capabilities during one run for activation on the next. CapabilityCreation exposes tools that let the model write an AbstractCapability subclass to disk as Python source and validate it immediately; the orchestrator loads active authored capabilities into the next agent.run(...).
CapabilityCreation is for capabilities authored by the agent as Python source. For capabilities written or selected by application code, see Building Custom Capabilities.
While Pydantic AI Harness is on 0.x releases, the API may change between minor releases; when it does, deprecation warnings and release-note migration guidance tell you (or your agent) exactly how to upgrade. See the version policy.
A coding agent often discovers, mid-task, that it wants a behavior its host does not yet have: a guardrail, an extra instruction, a tool, a request hook. The capability surface to express that already exists — but normally only a developer can write a capability class, wire it into the agent, and restart. Without runtime capability creation, the agent cannot author that extension during a run and make it available to the next run.
CapabilityCreation exposes three tools:
author_capability(name, code)— writecodeto<directory>/<name>.py, import it, and validate it. Validation requires exactly onepydantic_ai.capabilities.AbstractCapabilitysubclass that constructs with no arguments. It calls the static getters (get_instructions,get_toolset,get_native_tools,get_model_settings,get_serialization_name) and checks that the agent can use what they return. The async lifecycle hooks are not run — they need a liveRunContext.list_authored_capabilities()— list authored capabilities with their status and any validation error.disable_authored_capability(name)— mark a capability disabled, soload_active()skips it.
A “hook” is not a standalone object in pydantic-ai — it is a method on a capability. So authoring a hook means authoring a capability that overrides one lifecycle method. A single overridden hook is a valid capability.
Construct CapabilityCreation with a directory for the authored files, then add it to the agent’s capabilities. directory is a path on the machine running the agent, relative to the process’s current directory, not a workspace path.
from pathlib import Path
from pydantic_ai import Agent
from pydantic_ai.capabilities import LocalWorkspace
from pydantic_ai_harness import CapabilityCreation
creation = CapabilityCreation(directory=Path('.authored'))
agent = Agent('anthropic:claude-sonnet-5-5', capabilities=[LocalWorkspace('.'), creation])
The agent can now call author_capability, list_authored_capabilities, and disable_authored_capability. CapabilityCreation also contributes static, cache-stable system-prompt guidance explaining these tools. Leave guidance=None for the default text, or pass your own string; set guidance='' to omit it entirely.
Writing and validation happen in the current run; activation happens on the next run. Pydantic AI resolves the effective capability set once at the start of each run (the run’s root capability is fixed; there is no setter), so an authored capability’s tools and hooks only exist once the next run’s toolset and capability chain are assembled.
The successful author_capability result tells the model the capability does not take effect in the current run. Writing and validating a capability does not schedule or inject it automatically.
Pass creation.store.load_active() as capabilities= on each run; an authored capability is then active from the next run, without a process restart:
from pathlib import Path
from pydantic_ai import Agent
from pydantic_ai.capabilities import LocalWorkspace
from pydantic_ai_harness import CapabilityCreation
creation = CapabilityCreation(directory=Path('.authored'))
agent = Agent('anthropic:claude-sonnet-5-5', capabilities=[LocalWorkspace('.'), creation])
history = None
done = False
next_prompt = 'Start the task.'
while not done:
extra = creation.store.load_active()
result = await agent.run(next_prompt, message_history=history, capabilities=extra)
history = result.all_messages()
# ... decide `next_prompt` and `done` from `result` ...
creation.store is the disk-backed CapabilityStore over the same directory. store.load_active() re-imports every active authored capability and runs the same checks as author_capability before returning it for injection into the next run. An entry that fails (corrupt source, construction error, a getter with a wrong return type) is skipped rather than raised, and its error is recorded in the manifest, so one bad capability does not block the others. Each version of a capability that load_active() loads stays imported for the life of the process, so an instance from an earlier run keeps working after the model re-authors it.
Authored capabilities persist to disk: each is one <directory>/<name>.py file, indexed by a sibling manifest.json. A fresh process picks them up by constructing a new CapabilityCreation over the same directory and calling store.load_active().
manifest.json records each capability’s name, module file, class name, status (active or disabled), and last validation error. That is the surface a UI can read to show what the agent has authored. The manifest is written atomically (temp file plus os.replace), so a crash mid-write does not leave a partial file. A manifest that cannot be read (malformed JSON, or a field value this version does not accept) is read as empty, and author_capability and disable_authored_capability raise UserError instead of replacing it, so its entries are kept until you repair or remove it.
Every store over one directory in a process shares a lock around each manifest update, so agents in the same process do not lose each other’s updates. Separate processes are not coordinated: point each directory at one process at a time.
Capability names are at most 64 lowercase letters, digits, and underscores, starting with a letter. Reusing a name replaces the previous capability of that name once the new code passes validation. Code that fails is kept on disk so it can be inspected. If the name’s current entry has no recorded error (its last_error is unset) and its <name>.py file still exists, that version stays in place and the rejected code is saved as <name>.rejected.py. Otherwise the code is saved as <name>.py and recorded with its last_error set; load_active() skips it.
Under durable execution, each authoring, listing, and disabling call to the store is recorded, so a recovered run reuses the result instead of writing the authored module and manifest again. Temporal and Prefect record each tool call in its own activity or task. DBOS runs function tools in workflow code, so there the store call runs as its own step.
The records are named after the capability’s id, which defaults to
capability_creation, so durable execution needs no configuration.
Inside a Temporal or DBOS workflow, a run refuses capabilities passed with agent.run(..., capabilities=...), because those engines register a capability’s activities or steps when the capability is bound to the agent. The integration contract above therefore cannot activate authored capabilities there. A Prefect flow accepts an authored capability that only adds hooks, instructions, or model settings, and refuses one whose get_toolset() contributes a function, MCP, or dynamic toolset. The authoring tools work under all three.
CapabilityCreation imports model-written Python into the agent’s own process, on this
machine. A run therefore refuses to start unless its
workspace is a writable LocalWorkspace: next to a
sandbox the model’s code would run outside the sandbox, and a read-only workspace promises the
model that it changes nothing.
CapabilityCreation executes arbitrary Python in-process at import, construction, and run time. That is the same trust boundary an agent that already runs shell commands and edits files operates under. Do not point it at a directory whose contents you would not run yourself, and treat authored capabilities as code the agent is executing on your host.
Authored code first runs when the model calls author_capability, not on the next run: validation executes the module body, the constructor, and the static getters in the current run, with the agent process’s environment and permissions, even when validation then rejects the code. If that code blocks, cancelling the run stops waiting for it, but its worker thread runs until the code returns and then finishes the write, so code that passes validation is still saved and loaded by the next load_active(); disable it if you did not want it. Other authoring, listing, and disabling calls are not held up by it. store.load_active() runs authored code in its calling thread, so an authored module that blocks also blocks that call. Loads of the same source from different threads run one at a time, so it also holds up every other load of that source, such as a load_active() call in another thread.
To have a person approve authored code before any of it runs, require approval for author_capability with a ToolGuardrail. The run then ends with a DeferredToolRequests output for the call, which you approve or deny as described in Human-in-the-loop tool approval:
from pathlib import Path
from pydantic_ai import Agent, DeferredToolRequests
from pydantic_ai.capabilities import LocalWorkspace
from pydantic_ai_harness import CapabilityCreation, GuardrailResult, ToolGuardrail
agent = Agent(
'anthropic:claude-sonnet-5-5',
capabilities=[
LocalWorkspace('.'),
CapabilityCreation(directory=Path('.authored')),
ToolGuardrail(guard=lambda call: GuardrailResult.approve(), tools=['author_capability']),
],
output_type=[str, DeferredToolRequests],
)
If the agent also has file or shell tools (such as FileSystem or Shell) over a workspace that contains directory, as with LocalWorkspace('.') and directory=Path('.authored'), it can edit authored files and manifest.json directly, without going through author_capability. Keep directory outside that workspace if every authored file should pass through validation.
Because authored capabilities hold live code, they are not spec-serializable (get_serialization_name() returns None) and are persisted as source rather than as an agent spec.
CapabilityCreation emits no spans of its own. Core’s tool spans record each author_capability, list_authored_capabilities, and disable_authored_capability call, and authoring and validation run inside the author_capability span.
Bases: AbstractCapability[AgentDepsT]
Create Pydantic AI capabilities during one run for activation on the next.
Exposes author_capability(name, code), list_authored_capabilities, and
disable_authored_capability. Authoring validates the Python source (exactly
one AbstractCapability subclass that constructs with no arguments and whose
static getters run) before it replaces <name>.py in directory; code that
fails is kept for inspection and, while a working version exists, does not
replace it. Authored capabilities hold live code, so they are persisted as source
rather than as a spec.
Pydantic AI resolves the capability set once at the start of each run, so an
authored capability becomes usable on the next run: pass the store’s active
capabilities as capabilities=.
from pathlib import Path
from pydantic_ai import Agent
from pydantic_ai.capabilities import LocalWorkspace
from pydantic_ai_harness.capability_creation import CapabilityCreation
creation = CapabilityCreation(directory=Path('.authored'))
agent = Agent('anthropic:claude-sonnet-4-6', capabilities=[LocalWorkspace('.'), creation])
# Inject whatever the agent has authored so far.
result = await agent.run('build a logging capability', capabilities=creation.store.load_active())
Authoring imports model-written Python into the agent’s own process, on this
machine. So a run refuses to start unless its workspace is this machine
(LocalWorkspace) and writable: next to a sandbox, the model’s code would
escape the sandbox, and a read-only workspace promises the model changes
nothing.
Each store read and write runs as a durable operation, so under durable execution a recovered run reuses the recorded result instead of writing the authored module and manifest again.
Directory holding the authored <name>.py files and the manifest.json index.
A path on the machine running the agent, relative to the process’s current directory, not a workspace path.
Type: Path
Static system-prompt guidance on authoring. Cache-stable. Leave None for the
default, or set '' to omit guidance entirely.
Type: str | None Default: None
Stable identity for durable execution, which records each store read and write under it.
Type: str | None Default: 'capability_creation'
The disk-backed store. Call store.load_active() to inject authored capabilities into the next run.
Type: CapabilityStore
@classmethod
def combine(
cls,
capabilities: Sequence[AbstractCapability[AgentDepsT]],
) -> AbstractCapability[AgentDepsT]
Two under one id are one configuration stated twice; two that disagree raise rather than merge.
AbstractCapability[AgentDepsT]
@async
def before_run(ctx: RunContext[AgentDepsT]) -> None
Refuse the run unless its workspace is this machine and writable.
def get_instructions() -> AgentInstructions[AgentDepsT] | None
Static, cache-stable guidance on the authoring tools.
AgentInstructions[AgentDepsT] | None
def get_toolset() -> AgentToolset[AgentDepsT] | None
Toolset providing the authoring tools over this capability’s store.
AgentToolset[AgentDepsT] | None
@classmethod
def get_serialization_name(cls) -> str | None
Not spec-serializable: the capability holds a live, disk-backed store.
Read/write index of authored capability .py files under directory.
def write(name: str, code: str) -> AuthoredCapability
Validate code as the capability name and save it as <name>.py.
Raises ValueError for an invalid name and UserError for an unreadable
manifest, both before writing anything.
The code is validated before it replaces anything. Code that passes
replaces <name>.py and its manifest entry. Code that fails is kept on
disk so it can be inspected: when the current entry for name has no
recorded error and its <name>.py still exists, that version and its
manifest entry stay as they are, the rejected code is saved as
<name>.rejected.py, and the returned record (with that module_file)
is not saved to the manifest. Otherwise the code is saved as <name>.py
and recorded with last_error set, so load_active skips it.
If saving the manifest fails, the previous <name>.py (or its absence)
is put back before the error is raised, so the source and its manifest
entry change together or not at all.
AuthoredCapability
def disable(name: str) -> bool
Mark the named capability disabled so load_active stops returning it. Returns whether it existed.
def list_all() -> list[AuthoredCapability]
Return every manifest entry, in insertion order.
list[AuthoredCapability]
def load_active() -> list[AbstractCapability[object]]
Construct every active authored capability for per-run injection.
Re-imports, re-constructs, and re-validates each active entry with the same
checks as write. Entries that fail (corrupt source, construction error, a
getter with a wrong return type, a module file outside directory) are
skipped, not raised. A load outcome that disagrees with the record’s
last_error is persisted back to the manifest: a newly broken entry records
its error, a re-fixed entry clears it, so the manifest stays truthful about
which capabilities are actually active. An outcome is not persisted when the
entry or its source was rewritten while it was loading, keeping the newer
write’s state.
The returned instances are built from each file as this call read it. A
write that lands during the call may be in them or not, and the next call
picks it up. Making them always match the manifest would mean holding the
directory lock while authored code runs, which would let one module that
blocks hold up every write.