Agent¶

Learn: Your first agent. Concepts: Agent.

Agent ¶

Agent(model: str | Model, *, system: str | None = None, tools: Iterable[Any] = (), skills: Iterable[str] = (), loop: Any = None, reporter: Reporter | None = DEFAULT, human: Human | None = DEFAULT, store: Store | None = None, permissions: Iterable[Permission] | None = None, name: str | None = None, description: str | None = None)

An agent's settings, and the actions that need a model, tools or a human.

Settings do not change after creation. Use copy to get an Agent with different settings. The history of a run lives in a State, and the order of steps is decided by the loop.

Example

agent = Agent(model="claude-sonnet-5", system="You are a coding assistant", tools=[read_file])
answer = agent.run("Find the bug in main.py")

Creates an Agent. Mistakes in the settings are reported here, with how to fix them.

Creating an Agent uses no network and needs no API key.

Parameters:

Name	Type	Description	Default
`model`	`str \| Model`	A model name such as `"claude-sonnet-5"`, `"anthropic/claude-sonnet-5"` or `"ollama/llama3:8b"`, or a `Model` object such as `Anthropic(...)` when you need settings.	required
`system`	`str \| None`	The system prompt.	`None`
`tools`	`Iterable[Any]`	`@tool` functions, objects with `@tool` methods, single `@tool` methods, objects of `Tool` subclasses, `MCP` servers and single MCP tools (`github.create_issue`). An `Agent` with a `name` and a `description` becomes a subagent: the model sees a tool of that name and description that takes a `prompt`, and calling it runs that Agent on the prompt alone. The model gets the Agent's answer, with a note when it stopped at its limit or by a permission, or `(NAME failed: ...)` when it raised an exception after it started. A mistake found before the subagent starts (`NoHumanError`, a `TypeError` for something async-only in a sync run) is raised instead and stops the run. In a sync run, Ctrl+C stops the run but not a subagent that is running in a worker thread: it goes on until it ends and its result is recorded as a late result. Use `arun` to cancel subagents together with the run.	`()`
`skills`	`Iterable[str]`	Not implemented yet. A non-empty value raises `NotImplementedError`.	`()`
`loop`	`Any`	The loop `run` calls with `(agent, state)`. Defaults to `default_loop`.	`None`
`reporter`	`Reporter \| None`	Receives progress notifications. Defaults to the shared `Terminal`. `None` is silent. An Agent run as a subagent (inside a tool call of another run) that was not given one uses the reporter of the run that started it; `None` given on purpose stays silent.	`DEFAULT`
`human`	`Human \| None`	Answers `ask_human`. Defaults to the shared `Terminal`. `None` means no human. A subagent that was not given one asks the human of the run that started it.	`DEFAULT`
`store`	`Store \| None`	Saves the States this Agent runs as they go, so they can be continued with `store.load(id)`. `None` (the default) saves nothing, except for a subagent, which is saved in the store of the run that started it. A subagent's State is saved under its own id.	`None`
`permissions`	`Iterable[Permission] \| None`	Permissions that decide, before a tool call runs, whether it may run (see `alpineagents.permissions`). `None` (the default) runs every call without checks; to turn checks off on a copy, `agent.copy(permissions=None)`. A list, even an empty one, is asked about every call, and a call no permission allows is denied with a `PermissionWarning` (so `[]` denies every call). They apply to every tool call of the run, the calls inside its subagents included. A subagent without `permissions=` is checked by the same list, and a subagent with its own list is checked by that list, after the `DenyPermission` of the runs above it: those cannot be widened, but the allow and decide permissions of the parent are not asked about the subagent's calls (so the parent's "deny the rest" does not bind it). A permission asks the `human` of the Agent that wrote it. Put limits that must always hold in `DenyByName` and the like: they also bind another Agent that runs tools or starts a run inside a subagent's run (`other.use_tools(state)` or `other.run(...)` in a custom loop), which is checked by the `DenyPermission` of the runs above first, then by its own list.	`None`
`name`	`str \| None`	The Agent's name.	`None`
`description`	`str \| None`	What the Agent does.	`None`

Raises:

Type Description

TypeError

A setting has the wrong type, for example a function without @tool in tools=, a loop that is not callable, a Reporter, Human, Store or Permission class instead of an object. In tools=: an unbound method (FileSystem.read_file: create an object first), a Tool subclass instead of an object, a Tool object whose class does not call Tool.__init__ or does not define run, or an object with no @tool methods, or an Agent whose name is not a valid tool name.

ValueError

Two tools share a name, an Agent in tools= has no name or no description, or the model string is ambiguous.

model `property` ¶

model: Model

The Model object. A model string given to Agent(model=...) is already turned into one.

system `property` ¶

system: str | None

The system prompt, or None.

tools `property` ¶

tools: tuple[Any, ...]

The items passed in tools=, as given. tool_map has the tools they stand for, by name.

tool_map `property` ¶

tool_map: Mapping[str, Tool | MCPTool]

Every tool the model can call, by the name it calls it: @tool functions, the @tool methods of objects in tools=, and the tools of MCP servers (MCPTool, named {server}__{tool}).

Use it to look up the tool of a call, and read what the tool does from its hints (read_only, destructive, idempotent, open_world). Read-only: to change the tools, use copy(tools=...). MCP servers' tools are listed only while the servers are connected: during a run, or inside with agent:. Looking up a missing name before that raises a KeyError that says so.

Example

for call in state.pending_calls:
    tool = agent.tool_map.get(call.name)  # None for a name the model made up
    if tool is not None and not tool.read_only:
        print(f"{call.name} may change something")

loop `property` ¶

loop: Any

The loop run calls. default_loop unless loop= was given.

reporter `property` ¶

reporter: Reporter | None

The Reporter that receives progress notifications, or None. Defaults to the shared Terminal. This is the Agent's own setting: a subagent that was not given one reports to the Reporter of the run that started it (see Agent(reporter=)), and this property does not show that.

human `property` ¶

human: Human | None

The Human that answers ask_human, or None. Defaults to the shared Terminal. This is the Agent's own setting: a subagent that was not given one asks the Human of the run that started it (see Agent(human=)), and this property does not show that.

store `property` ¶

store: Store | None

The Store that saves this Agent's States as they run, or None. This is the Agent's own setting: a subagent without one is saved in the Store of the run that started it, and this property does not show that.

permissions `property` ¶

permissions: tuple[Permission, ...]

The permissions that decide whether each tool call may run, in the order given. Empty both without permissions= (every call runs unchecked) and with an empty list (every call is denied).

name `property` ¶

name: str | None

The Agent's name, or None.

description `property` ¶

description: str | None

What the Agent does, or None.

run ¶

run(prompt_or_state: str | State) -> Any

Runs the loop on a prompt or a State and returns what the loop returns, usually the answer.

A prompt string starts a new State with that one user message. Any Agent can continue any State, so agent.copy(model=...) can carry on where another model left off. A State is run by one run or arun at a time.

MCP servers in tools= connect for the run and disconnect after it, unless with agent: keeps them open. If an exception leaves the run, it is recorded in state.history, tool calls still waiting for a result are closed ((interrupted by user) after Ctrl+C), and the exception is raised as is. After Ctrl+C you can call run again with the same State.

Closed calls get (interrupted by user) after Ctrl+C or cancellation and (aborted: TimeoutError) (the exception's type) after another exception, SystemExit included (so turning SIGTERM into SystemExit ends a run cleanly). A sync tool still running keeps running in its thread; if it finishes later, the model gets its result as a notice at the next think. Afterwards state.stopped is None.

Continuing a State that already ran: if its last run stopped at an until function that is still true, the loop stops at once (add a message first). If it stopped at limit, it runs up to limit more turns, because each call counts turns from zero. A State ended with finish() raises ValueError.

With store=, the State is saved as the run goes: at the start and end of the run, before and after each think, after each tool result, and after use_tools and compact. What permission checks recorded (denied calls, the person's answers) is saved before any tool of the turn runs. A save also happens after an exception. If saving fails during a run that goes well, the run raises the store's exception. If the run is already failing, the original exception is raised with a note about the failed save, so except RateLimitError: still catches it; Ctrl+C or cancellation while saving is raised instead of the original. A failed save after one tool result does not stop the other tools: the save at the end of use_tools writes what is missing, or raises. Nothing that failed to save is lost: the next save writes it.

Parameters:

Name	Type	Description	Default
`prompt_or_state`	`str \| State`	A prompt string, which starts a new `State`, or a `State` to continue.	required

Returns:

Type	Description
`Any`	What the loop returns. The default loop returns `state.answer`.

Raises:

Type	Description
`TypeError`	The loop is async (use `arun`), the argument is neither a string nor a State, a value in the State cannot be saved as JSON, the store only works asynchronously, or a permission only checks asynchronously (implements `acheck` but not `check`).
`NoHumanError`	A `DecideByHuman()` permission would ask the Agent's human, and the Agent has `human=None`.
`ValueError`	The prompt is empty, the State was already ended with `state.finish()`, another `run`/`arun` is running it right now, it is new and the store already has a State with its id, it is saved in another store (a State is saved in one store only; two `FileStore` objects for the same folder count as one), or an MCP server has no tool that `tools=` picked, or two of its tools get the same name (raised when the server connects, before the first `think`).
`TypeError`	Also when called inside an async run, where it would block the event loop (use `arun`).

Warns:

Type	Description
`ResumeWarning`	The State was loaded from a store, and it was last run by an Agent with a different name, model, system prompt, tools or MCP servers. Only the first run after `store.load` compares.

Example

state = State(messages=[Message.user("Find the bug in this repo")])
answer = agent.run(state)
print(state.stopped, state.usage.cost)

arun `async` ¶

arun(prompt_or_state: str | State) -> Any

The async version of run.

Uses this Agent's loop if it is async, or adefault_loop if the Agent uses the default loop. Cancelling the task follows the same rules as Ctrl+C in run.

Parameters:

Name	Type	Description	Default
`prompt_or_state`	`str \| State`	A prompt string, which starts a new `State`, or a `State` to continue.	required

Returns:

Type	Description
`Any`	What the loop returns. The default loop returns `state.answer`.

Raises:

Type	Description
`TypeError`	The loop is a sync loop other than `default_loop`, which would block the event loop.
`ValueError`	The prompt is empty, the State was already ended with `state.finish()`, or another `run`/`arun` is running it right now.
`NoHumanError`	A `DecideByHuman()` permission would ask the Agent's human, and the Agent has `human=None`.

Warns:

Type	Description
`ResumeWarning`	The State was loaded from a store, and it was last run by an Agent with a different name, model, system prompt, tools or MCP servers. Only the first run after `store.load` compares.

think ¶

think(state: State, tools: Iterable[Any] | None = None) -> None

Sends the messages to the model once and records its reply in the State.

The reply may ask for tool calls. Check state.pending_calls and run them with use_tools. If the model call fails, the request is taken back (state.messages and state.turn are as they were before this call) and the error is raised and recorded in the history. Once the reply is recorded, nothing undoes it.

Any Agent can think on any State, so a State can be continued with another model.

Parameters:

Name	Type	Description	Default
`state`	`State`	The State to continue.	required
`tools`	`Iterable[Any] \| None`	Which of this Agent's tools the model may see this turn. `None` shows all of them, `[]` shows none. Takes the items of `tools=` and the values of `tool_map`, so `[t for t in agent.tool_map.values() if t.read_only]` works.	`None`

Raises:

Type	Description
`ValueError`	The State is finished, has no messages, still has tool calls waiting for results, is already waiting on a model, or `tools` has a tool this Agent does not have.
`TypeError`	Called inside an async run, where it would block the event loop (use `athink`).
`ProviderError`	The model provider failed (`RateLimitError`, `AuthError`, `ContextTooLongError` or another `ProviderError`).

athink `async` ¶

athink(state: State, tools: Iterable[Any] | None = None) -> None

The async version of think. Cancelling it takes the request back like any other failure.

use_tools ¶

use_tools(state: State) -> None

Runs the tool calls from the last reply and records their results in the State.

Calls to tools made with parallel=True (the default) run at the same time, then the rest run one by one. Does nothing if there are no calls. A tool that raises does not become an error result: results of the calls that finished are recorded, and the exception is raised as is, with a note naming the call. The calls that did not finish stay pending, so the next think raises ValueError: call use_tools again (calls that already have a result do not run again) or let the exception leave the run, which closes them. Invalid arguments and unknown tool names go back to the model as (input error: ...) results.

The calls run in this order: tools with parallel=True on worker threads at the same time, then tools with parallel=False one at a time in the order the model asked. The results go into state.messages together, in the order the model asked, once the last call has one.

With permissions=, every call is checked before any tool runs. A denied call does not run, and the model gets the reason as its error result. A denial with stop=True cancels the other calls of the turn too, so no tool of the turn runs, and the loop stops before its next turn. Called on the State of a subagent's run by an Agent that is not that run's own (a custom loop calling other.use_tools(state)), the call is checked by the DenyPermission of the runs above the subagent first, then by this Agent's list.

Every call ends as one ToolResultEntry whose outcome says how: done, error, input_error, aborted, interrupted, denied or cancelled.

Parameters:

Name	Type	Description	Default
`state`	`State`	The State whose `pending_calls` to run.	required

Raises:

Type	Description
`ValueError`	The State is finished.
`TypeError`	Called inside an async run, where it would block the event loop (use `ause_tools`), or a permission only checks asynchronously.

ause_tools `async` ¶

ause_tools(state: State) -> None

The async version of use_tools, with the same checks: inside a subagent's run, an Agent that is not that run's own is checked by the DenyPermission of the runs above first.

async def tools run as tasks on the running event loop, other tools on worker threads. Cancelling it follows the same rules as Ctrl+C.

ask ¶

ask(state: State, prompt: str, returns: Any = str, *, retries: int = 2) -> Any

Asks the model a question about the current context and returns the answer in the returns format.

The question and answer are recorded in state.history but do not enter the context, so the run continues as if the question was never asked. The model cannot call tools while answering. Any Agent can ask about a State, not only the one running it. It adds nothing to state.turn and writes no model_request entry, but its tokens count in state.usage.

Parameters:

Name	Type	Description	Default
`state`	`State`	The State whose context the model reads.	required
`prompt`	`str`	The question.	required
`returns`	`Any`	The answer format: `str`, a dataclass or a Pydantic model.	`str`
`retries`	`int`	How many more times to ask when the answer does not fit `returns` (2 by default).	`2`

Returns:

Type	Description
`Any`	The answer, converted to `returns`.

Raises:

Type	Description
`TypeError`	`returns` is not a supported format, or called inside an async run (use `aask`).
`ValueError`	The State is finished (so do not combine `ask` with a `submit` tool that calls `finish` on the same State).
`OutputError`	The answer still did not fit `returns` after `retries` more tries.
`ProviderError`	The model provider failed.

Example

@dataclass
class Review:
    approved: bool
    reason: str

review = agent.ask(state, "Is this change safe to merge?", returns=Review)

aask `async` ¶

aask(state: State, prompt: str, returns: Any = str, *, retries: int = 2) -> Any

The async version of ask.

ask_human ¶

ask_human(state: State, prompt: str, returns: Any = str) -> Any

Asks the person through human and returns the answer in the returns format.

The question and answer are recorded in state.history. The context does not change.

Parameters:

Name	Type	Description	Default
`state`	`State`	The State of the current run.	required
`prompt`	`str`	The question.	required
`returns`	`Any`	The answer format: `str`, `bool` or `Literal[...]` for a fixed set of choices.	`str`

Returns:

Type	Description
`Any`	The answer, converted to `returns`.

Raises:

Type	Description
`NoHumanError`	The Agent was created with `human=None`.
`TypeError`	`returns` is not a supported format, the Human only answers asynchronously (use `aask_human`), or called inside an async run (use `aask_human`).

Example

choice = agent.ask_human(state, "Run the command?", returns=Literal["yes", "no", "always"])

aask_human `async` ¶

aask_human(state: State, prompt: str, returns: Any = str) -> Any

The async version of ask_human.

Awaits human.aask when the Human has it, and otherwise runs human.ask on a worker thread.

compact ¶

compact(state: State, instructions: str | None = None) -> None

Asks the model to summarize the messages, then replaces them with the first user message and that summary.

It is the same change as state.compact(summary) with the model's summary, and the usage of the request is added to state.usage. Like ask, it writes no model_request entry, so the model that wrote a summary is not in history. Messages added while the model was writing the summary stay after it. state.history keeps everything. If anything fails, the messages are left unchanged. In a loop, compact_if_full calls this only when the context is getting full.

Parameters:

Name	Type	Description	Default
`state`	`State`	The State to compact.	required
`instructions`	`str \| None`	What the summary should keep, for example `"Keep file paths and failing tests"`.	`None`

Raises:

Type	Description
`ValueError`	The State still has tool calls waiting for results, or another compaction of it is under way.
`TypeError`	Called inside an async run, where it would block the event loop (use `acompact`; in a loop, `acompact_if_full`). `compact_if_full` raises this only when it first compacts, which can be many turns in.
`OutputError`	The model returned an empty summary.
`ProviderError`	The model provider failed.

acompact `async` ¶

acompact(state: State, instructions: str | None = None) -> None

The async version of compact.

context_tokens ¶

context_tokens(state: State) -> int

Estimated size of what the model would receive next, in tokens: the State's messages plus this Agent's system prompt and tool definitions.

An estimate: it starts from the token count the provider reported for the last reply and adds a character-count guess for the messages after it. Each image, in a user message or a tool result, counts as 1,600 tokens.

Example

print(f"{agent.context_tokens(state):,} tokens")

context_used ¶

context_used(state: State) -> float

The fraction of this Agent's model's context window that context_tokens(state) fills. Can go above 1.0. compact_if_full compares it with its at.

Example

if agent.context_used(state) > 0.8:
    agent.compact(state)

copy ¶

copy(**changes: Any) -> Agent

Returns a new Agent with some settings changed. The original Agent does not change.

The new Agent goes through the same checks as Agent(...).

Parameters:

Name	Type	Description	Default
`**changes`	`Any`	Settings to change, by the same names as the `Agent(...)` arguments.	`{}`

Returns:

Type	Description
`Agent`	The new Agent.

Raises:

Type	Description
`TypeError`	A name in `changes` is not a setting.

Example

quiet = agent.copy(model=FakeModel(["Done"]), reporter=None)

Agent¶

Agent ¶

model property ¶

system property ¶

tools property ¶

tool_map property ¶

loop property ¶

reporter property ¶

human property ¶

store property ¶

permissions property ¶

name property ¶

description property ¶

run ¶

arun async ¶

think ¶

athink async ¶

use_tools ¶

ause_tools async ¶

ask ¶

aask async ¶

ask_human ¶

aask_human async ¶

compact ¶

acompact async ¶

context_tokens ¶

context_used ¶

copy ¶

model `property` ¶

system `property` ¶

tools `property` ¶

tool_map `property` ¶

loop `property` ¶

reporter `property` ¶

human `property` ¶

store `property` ¶

permissions `property` ¶

name `property` ¶

description `property` ¶

arun `async` ¶

athink `async` ¶

ause_tools `async` ¶

aask `async` ¶

aask_human `async` ¶

acompact `async` ¶