#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

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

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

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

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

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

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

Build Tool objects from ToolSchemas + handlers, or None.

#_resolve_model

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

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

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

#invoke_agent

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

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.