claudestream v0.14.2 /claudestream._agent
Updated
On this page

Agent definition loader and budget enforcement for Claude Code sessions, with sync and async context managers for invoking agents.

#claudestream._agent

#claudestream._agent

Agent definition loader and budget enforcement for Claude Code sessions, with sync and async context managers for invoking agents.

#AgentValidationError

Raised when an at-rest .agent.json document fails strictspec schema validation (the integer format_version gate, closed-record unknown-key rejection, or a per-field type check).

Guards the AT-REST boundary only. In-memory :class:AgentDefinition construction is never routed through the schema.

#AgentDefinition

A complete agent definition, loadable from a .agent.json file.

#_check_deprecated_budget

python
def _check_deprecated_budget(data: bytes) -> None

Raise a migration hint for a document still using the old budget fields.

Runs BEFORE the structural schema gate: the budget rename has no live strictspec migration (it ships only as a conformance fixture), so without this the reader would get a generic unknown-key diagnostic instead of the replacement field names.

#_decode_agent_document

python
def _decode_agent_document(data: bytes, source: str) -> AgentDefinition

Validate raw .agent.json bytes and decode into an :class:AgentDefinition.

This is the single at-rest boundary: every entry point that reads a .agent.json from disk or a package goes through it, so a given document produces the same diagnostic no matter how it was reached.

Order matters. The claudestream-owned deprecated-budget hint runs first, because the schema would otherwise reject those fields as unknown keys with no remediation. Then the strictspec gate: a missing/wrong-typed integer format_version, an unknown key, or a wrong-typed field is a hard error (:class:AgentValidationError) carrying the pinned diagnostic code, path, and remediation text. msgspec decode runs only on a document that already passed.

#resolve_prompt

python
def resolve_prompt(template: str, variables: dict[str, str]) -> str

Resolve {variable} placeholders in a prompt template.

Only placeholders present in the original template are considered template variables. Curly-brace patterns introduced by substituted values (e.g. {rects} inside a TypeScript API reference) are left as-is and do not trigger validation errors.

Raises:

  • ValueError: If any original template placeholders remain after

substitution (i.e. the caller forgot to supply a variable).

#load_agent

python
def load_agent(path: str | Path, cwd: str | None=None) -> AgentDefinition

Load an AgentDefinition from a .agent.json file or by bare name.

If path has no path separators and doesn't end with .json, it is treated as a bare agent name. The loader looks for .claudestream/agents/<name>.agent.json relative to cwd (or the current working directory when cwd is None).

#discover_agents

python
def discover_agents(cwd: str | None=None, paths: list[str] | None=None, packages: list[str] | None=None) -> list[AgentDefinition]

Discover agent definitions from multiple sources.

Sources are searched in order; the first occurrence of each agent name wins.

  1. .claudestream/agents/ relative to cwd (or the current working

directory when cwd is None).

  1. Each directory in paths (relative paths resolved against cwd).
  2. Each Python package in packages via importlib.resources.

Returns a deduplicated list of :class:AgentDefinition sorted by name.

#_build_tools

python
def _build_tools(definition: AgentDefinition, tool_handlers: dict[str, Any] | None) -> list | None

Build Tool objects from ToolSchemas + handlers, or None.

#_resolve_model

python
def _resolve_model(config: SessionConfig, definition: AgentDefinition) -> str

Return the effective model: definition wins if set, then config.

Raises:

  • ValueError: If neither source provides a model.

#_build_session_resolution

python
def _build_session_resolution(definition: AgentDefinition) -> SessionResolution | None

Build a SessionResolution from the agent name, or None if no name.

#invoke_agent

python
async def invoke_agent(definition: AgentDefinition, config: SessionConfig, *, variables: dict[str, str] | None=None, tool_handlers: dict[str, Any] | None=None)

Create and manage an AsyncSession from an AgentDefinition.

Uses config as the base configuration. Definition fields (model, sandbox, mcp, stream, system_prompt) override the config where set.

Raises:

  • ValueError: If model is not specified in the definition or config.
  • ValueError: If prompt template has unresolved variables.

#invoke_agent_sync

python
def invoke_agent_sync(definition: AgentDefinition, config: SessionConfig, *, variables: dict[str, str] | None=None, tool_handlers: dict[str, Any] | None=None)

Create and manage a SyncSession from an AgentDefinition.

Sync version of invoke_agent. Uses config as the base configuration. Definition fields override the config where set.

Raises:

  • ValueError: If model is not specified in the definition or config.
  • ValueError: If prompt template has unresolved variables.
Search