fast-agent Class Reference

This document provides detailed reference information for programmatically using the FastAgent class, which is the core class for creating and running agent applications.

FastAgent Class

Constructor

FastAgent(
    name: str,
    config_path: str | None = None,
    ignore_unknown_args: bool = False,
    parse_cli_args: bool = True,
    quiet: bool = False,
    skills_directory: str | Path | None = None,
    **kwargs,
)

Parameters

Parameter	Type	Default	Description
`name`	`str`	(required)	Name of the application
`config_path`	`str \\| None`	`None`	Optional path to config file. If not provided, config is loaded from default locations
`ignore_unknown_args`	`bool`	`False`	Whether to ignore unknown command line arguments when `parse_cli_args` is `True`
`parse_cli_args`	`bool`	`True`	Whether to parse command line arguments. Set to `False` when embedding FastAgent in frameworks like FastAPI/Uvicorn that handle their own argument parsing
`quiet`	`bool`	`False`	Disable progress display, tool and message logging for cleaner output
`skills_directory`	`str \\| Path \\| None`	`None`	Override the default skills directory
`**kwargs`	`Any`		Additional keyword args (advanced use)

Decorator Methods

The FastAgent class provides several decorators for creating agents and workflows:

Decorator	Description
`@fast.agent()`	Create a basic agent
`@fast.chain()`	Create a chain workflow
`@fast.router()`	Create a router workflow
`@fast.parallel()`	Create a parallel workflow
`@fast.evaluator_optimizer()`	Create an evaluator-optimizer workflow
`@fast.orchestrator()`	Create an orchestrator workflow
`@fast.iterative_planner()`	Create an iterative planner workflow
`@fast.maker()`	Create a MAKER (k-voting) workflow

See Defining Agents for detailed usage of these decorators.

Workflow Decorators (Generated)

These signatures are generated from the installed fast_agent package to prevent drift.

`chain`

fast.chain(
    name: str,
    *,
    sequence: list[str],
    instruction: str | pathlib.Path | pydantic.networks.AnyUrl | None = None,
    cumulative: bool = False,
    default: bool = False
) -> Callable[
    [Callable[~P, Coroutine[Any, Any, +R]]],
    Callable[~P, Coroutine[Any, Any, +R]],
]

`parallel`

fast.parallel(
    name: str,
    *,
    fan_out: list[str],
    fan_in: str | None = None,
    instruction: str | pathlib.Path | pydantic.networks.AnyUrl | None = None,
    include_request: bool = True,
    default: bool = False
) -> Callable[
    [Callable[~P, Coroutine[Any, Any, +R]]],
    Callable[~P, Coroutine[Any, Any, +R]],
]

`evaluator_optimizer`

fast.evaluator_optimizer(
    name: str,
    *,
    generator: str,
    evaluator: str,
    instruction: str | pathlib.Path | pydantic.networks.AnyUrl | None = None,
    min_rating: str = 'GOOD',
    max_refinements: int = 3,
    refinement_instruction: str | None = None,
    default: bool = False
) -> Callable[
    [Callable[~P, Coroutine[Any, Any, +R]]],
    Callable[~P, Coroutine[Any, Any, +R]],
]

`router`

fast.router(
    name: str,
    *,
    agents: list[str],
    instruction: str | pathlib.Path | pydantic.networks.AnyUrl | None = None,
    servers: list[str] | None = None,
    tools: dict[str, list[str]] | None = None,
    resources: dict[str, list[str]] | None = None,
    prompts: dict[str, list[str]] | None = None,
    model: str | None = None,
    use_history: bool = False,
    save_trajectory: bool = False,
    request_params: fast_agent.llm.request_params.RequestParams | None = None,
    human_input: bool = False,
    default: bool = False,
    elicitation_handler: mcp.client.session.ElicitationFnT | None = None,
    api_key: str | None = None
) -> Callable[
    [Callable[~P, Coroutine[Any, Any, +R]]],
    Callable[~P, Coroutine[Any, Any, +R]],
]

`orchestrator`

fast.orchestrator(
    name: str,
    *,
    agents: list[str],
    instruction: str | pathlib.Path | pydantic.networks.AnyUrl = '...',
    model: str | None = None,
    request_params: fast_agent.llm.request_params.RequestParams | None = None,
    use_history: bool = False,
    save_trajectory: bool = False,
    human_input: bool = False,
    plan_type: Literal['full', 'iterative'] = 'full',
    plan_iterations: int = 5,
    default: bool = False,
    api_key: str | None = None
) -> Callable[
    [Callable[~P, Coroutine[Any, Any, +R]]],
    Callable[~P, Coroutine[Any, Any, +R]],
]

`iterative_planner`

fast.iterative_planner(
    name: str,
    *,
    agents: list[str],
    instruction: str | pathlib.Path | pydantic.networks.AnyUrl = '...',
    model: str | None = None,
    request_params: fast_agent.llm.request_params.RequestParams | None = None,
    plan_iterations: int = -1,
    default: bool = False,
    api_key: str | None = None
) -> Callable[
    [Callable[~P, Coroutine[Any, Any, +R]]],
    Callable[~P, Coroutine[Any, Any, +R]],
]

`maker`

fast.maker(
    name: str,
    *,
    worker: str,
    k: int = 3,
    max_samples: int = 50,
    match_strategy: str = 'exact',
    red_flag_max_length: int | None = None,
    instruction: str | pathlib.Path | pydantic.networks.AnyUrl | None = None,
    default: bool = False
) -> Callable[
    [Callable[~P, Coroutine[Any, Any, +R]]],
    Callable[~P, Coroutine[Any, Any, +R]],
]

Methods

`run()`

async with fast.run() as agent:
    # Use agent here

An async context manager that initializes all registered agents and returns an AgentApp instance that can be used to interact with the agents.

`harness()`

fast.harness(*, model: str | None = None) -> AgentHarness

Creates a headless AgentHarness for typed, session-oriented Python usage. The harness uses the same initialization path as run() but does not enter the TUI, CLI message/prompt-file modes, MCP server mode, or ACP server mode. On startup, it loads AgentCards from the active environment's agent-cards/ directory when that directory exists and contains cards.

Parameter	Type	Default	Description
`model`	`str \\| None`	`None`	Optional global model override, similar to the CLI `--model` override

`start_server()`

await fast.start_server(
    transport: str = "http",
    host: str = "127.0.0.1",
    port: int = 8000,
    server_name: str | None = None,
    server_description: str | None = None,
    tool_description: str | None = None,
    instance_scope: str | None = None,
    permissions_enabled: bool = True,
    tool_name_template: str | None = None,
)

Starts the application as an MCP server.

Parameter	Type	Default	Description
`transport`	`str`	`"http"`	Transport protocol to use (`http`, `stdio`, `acp`, `a2a`)
`host`	`str`	`"127.0.0.1"`	Host address for the server when using HTTP
`port`	`int`	`8000`	Port for the server when using HTTP
`server_name`	`str \\| None`	`None`	Optional custom name for the MCP server
`server_description`	`str \\| None`	`None`	Optional description for the MCP server
`tool_description`	`str \\| None`	`None`	Customise the exposed `send` tool description (supports `{agent}` placeholder)
`instance_scope`	`str \\| None`	`None`	Control how clients receive isolated agent instances (`shared`, `connection`, `request`); `None` uses the transport default
`permissions_enabled`	`bool`	`True`	Enable tool permission requests (ACP only)
`tool_name_template`	`str \\| None`	`None`	Customise exposed agent tool names (supports `{agent}` placeholder)

`main()`

is_server_mode = await fast.main()

Helper method for checking if server mode was requested for this app. Server mode is triggered by --transport when running from the CLI, but that check happens in run().

AgentHarness Class

AgentHarness is an async context manager returned by FastAgent.harness().

from fast_agent import AgentHarness


async with fast.harness() as harness:
    typed: AgentHarness = harness

Properties

Property	Type	Description
`sessions`	`HarnessSessions`	Session manager for the running harness

Methods

`session()`

await harness.session(session_id: str | None = None, *, agent_name: str | None = None) -> HarnessSession

Returns an existing session or creates one by delegating to harness.sessions.get_or_create().

`shell()`

await harness.shell(
    command: str,
    *,
    cwd: str | Path | None = None,
    env: Mapping[str, str] | None = None,
    timeout: float | None = None
) -> ShellExecutionResult

Runs a shell command through the harness shell executor and returns a ShellExecutionResult with stdout, stderr, and exit_code. This is programmatic shell access: it does not create a session and does not add the command or output to chat history.

HarnessSessions Class

Manager for harness sessions. When session_history is enabled, creating a session also creates or loads environment_dir/sessions/<session_id>/.

session = await harness.sessions.get("demo")
session = await harness.sessions.create("demo")
session = await harness.sessions.get_or_create("demo")
await harness.sessions.delete("demo")

Session ID rules:

None means "default";
strings are stripped;
empty strings raise ValueError;
IDs must be 1-128 characters;
IDs must start and end with a letter or digit;
IDs may contain only letters, digits, dashes, and underscores.

Generated from HARNESS_SESSION_ID_PATTERN: ^[A-Za-z0-9](?:[A-Za-z0-9_-]{0,126}[A-Za-z0-9])?$.

Method	Signature	Behavior
`get`	`get(session_id: str \\| None = None) -> HarnessSession`	Return an existing session. Raises if missing.
`create`	`create(session_id: str \\| None = None, *, agent_name: str \\| None = None) -> HarnessSession`	Create a session. Raises if it already exists.
`get_or_create`	`get_or_create(session_id: str \\| None = None, *, agent_name: str \\| None = None) -> HarnessSession`	Return an existing session or create it.
`delete`	`delete(session_id: str \\| None = None) -> None`	Delete a session if present; missing sessions are ignored.

HarnessSession Class

A stable session backed by one owned AgentInstance.

Properties

Property	Type	Description
`id`	`str`	Normalized session ID
`default_agent_name`	`str \\| None`	Session default agent used when calls omit `agent_name`

Methods

await session.send(
    message: MessageParam,
    *,
    agent_name: str | None = None,
    request_params: RequestParams | None = None
) -> str

await session.generate(
    messages: MessageParam,
    *,
    agent_name: str | None = None,
    request_params: RequestParams | None = None
) -> PromptMessageExtended

await session.structured(
    messages: MessageParam,
    model: type[ModelT],
    *,
    agent_name: str | None = None,
    request_params: RequestParams | None = None
) -> tuple[ModelT | None, PromptMessageExtended]

await session.structured_schema(
    messages: MessageParam,
    schema: dict[str, Any],
    *,
    agent_name: str | None = None,
    request_params: RequestParams | None = None
) -> tuple[Any | None, PromptMessageExtended]

await session.clear(*, agent_name: str | None = None, clear_prompts: bool = False) -> None

await session.shell(
    command: str,
    *,
    cwd: str | Path | None = None,
    env: Mapping[str, str] | None = None,
    timeout: float | None = None
) -> ShellExecutionResult

await session.delete() -> None

Calls resolve their target agent in this order:

explicit per-call agent_name;
the session default_agent_name;
the app default agent.

clear() clears only the resolved target agent, not every agent in the session.

shell() returns ShellExecutionResult with stdout, stderr, and exit_code. It does not add the command or output to chat history, but it is serialized with other operations on the same HarnessSession.

Session lifecycle and concurrency

the same session ID returns the same HarnessSession object and the same owned AgentInstance;
different session IDs receive isolated AgentInstance objects;
when session_history is enabled, session IDs map to persisted environment_dir/sessions/<session_id>/ directories and existing histories are hydrated on creation;
deleting a session disposes its instance;
deleting a session removes its persisted session folder when persistence is enabled;
exiting the harness context disposes all remaining session instances;
deleted HarnessSession objects are closed.

Concurrent operations on the same HarnessSession are rejected:

RuntimeError: Session 'support-123' is already running generate; start another session for parallel conversation branches.

Deleting an active session is also rejected:

RuntimeError: Session 'support-123' is running generate; wait before deleting it.

AgentApp Class

The AgentApp class is returned from fast.run() and provides access to all registered agents and their capabilities.

Accessing Agents

There are two ways to access agents in the AgentApp:

# Attribute access
response = await agent.agent_name.send("Hello")

# Dictionary access
response = await agent["agent_name"].send("Hello")

Methods

`send()`

await agent.send(
    message: str | PromptMessage | PromptMessageExtended,
    agent_name: str | None = None,
    request_params: RequestParams | None = None,
) -> str

Send a message to the specified agent (or the default agent if not specified).

`apply_prompt()`

await agent.apply_prompt(
    prompt: str | GetPromptResult,
    arguments: dict[str, str] | None = None,
    agent_name: str | None = None,
    as_template: bool = False,
) -> str

Apply a prompt template to an agent (default agent if not specified).

`with_resource()`

await agent.with_resource(
    prompt_content: str | PromptMessage | PromptMessageExtended,
    resource_uri: str,
    server_name: str | None = None,
    agent_name: str | None = None
) -> str

Send a message with an attached MCP resource.

`interactive()`

await agent.interactive(
    agent_name: str | None = None,
    default_prompt: str = "",
    pretty_print_parallel: bool = False,
    request_params: RequestParams | None = None,
) -> str

Start an interactive prompt session with the specified agent.

Example: Integrating with FastAPI

See here for more examples of using FastAPI with fast-agent.

fastapi-simple.py

from contextlib import asynccontextmanager

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel

from fast_agent.core.fastagent import FastAgent

# Create FastAgent without parsing CLI args (plays nice with uvicorn)
fast = FastAgent("fast-agent demo", parse_cli_args=False, quiet=True)


# Register a simple default agent via decorator
@fast.agent(name="helper", instruction="You are a helpful AI Agent.", default=True)
async def decorator():
    pass


# Keep FastAgent running for the app lifetime
@asynccontextmanager
async def lifespan(app: FastAPI):
    async with fast.run() as agents:
        app.state.agents = agents
        yield


app = FastAPI(lifespan=lifespan)


class AskRequest(BaseModel):
    message: str


class AskResponse(BaseModel):
    response: str


@app.post("/ask", response_model=AskResponse)
async def ask(req: AskRequest) -> AskResponse:
    try:
        result = await app.state.agents.send(req.message)
        return AskResponse(response=result)
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

Example: Embedding in a Command-Line Tool

Here's an example of embedding FastAgent in a custom command-line tool:

import asyncio
import argparse
import sys
from fast_agent.core.fastagent import FastAgent

# Parse our own arguments first
parser = argparse.ArgumentParser(description="Custom AI Tool")
parser.add_argument("--input", help="Input data for analysis")
parser.add_argument("--format", choices=["text", "json"], default="text", help="Output format")
args, remaining = parser.parse_known_args()

# Create FastAgent with parse_cli_args=False since we're handling our own args
fast = FastAgent("Embedded Agent", parse_cli_args=False)

@fast.agent(instruction="You are a data analysis assistant")
async def analyze():
    async with fast.run() as agent:
        if not args.input:
            print("Error: --input is required")
            sys.exit(1)

        result = await agent.send(f"Analyze this data: {args.input}")

        if args.format == "json":
            import json
            print(json.dumps({"result": result}))
        else:
            print(result)

if __name__ == "__main__":
    asyncio.run(analyze())

This example shows how to: 1. Parse your application's own arguments using argparse 2. Create a FastAgent instance with parse_cli_args=False 3. Use your own command-line arguments in combination with fast-agent