API Reference (auto-generated)

Generated from source docstrings

The signatures and docstrings on this page are rendered directly from the code by
mkdocstrings at build time. They cannot drift from the
implementation — when the code changes, this page changes with it. The hand-written pages
under API Reference (Protocols, Providers, Tools, Workflows) provide the prose,
guidance, and examples that complement these signatures.

Public framework API

The core abstractions exposed from victor.framework.

Agent

victor.framework.agent.Agent(orchestrator: 'AgentOrchestrator', provider: str = 'anthropic', model: Optional[str] = None, vertical: Optional[Type['VerticalBase']] = None, vertical_config: Optional['VerticalConfig'] = None)

Simplified interface for creating and using Victor agents.

The Agent class provides a "golden path" API that covers 90% of use cases
with minimal configuration. For advanced use cases, access the underlying
AgentOrchestrator via get_orchestrator().

Attributes:

Name	Type	Description
state	State	Observable agent state (stage, tool usage, files)

Example - Simple usage

agent = await Agent.create(provider="anthropic", model="claude-sonnet-4-20250514")
result = await agent.run("Write a function to parse JSON")
print(result.content)

Example - With tools

agent = await Agent.create(
provider="anthropic",
tools=["filesystem", "git"]
)
result = await agent.run("Create a new feature branch and add a README")

Example - Streaming with events

async for event in agent.stream("Refactor this file"):
if event.type == EventType.THINKING:
print(f"Thinking: {event.content}")
elif event.type == EventType.TOOL_CALL:
print(f"Tool: {event.tool_name}({event.arguments})")
elif event.type == EventType.CONTENT:
print(event.content, end="")

Example - State observation

agent.on_state_change(lambda old, new: print(f"{old} -> {new}"))

Example - Escape hatch to full power

orchestrator = agent.get_orchestrator()

Access all internal components

Initialize Agent with orchestrator. Use Agent.create() instead.

Parameters:

Name	Type	Description	Default
orchestrator	'AgentOrchestrator'	AgentOrchestrator instance	required
provider	str	Provider name for reference	'anthropic'
model	Optional[str]	Model name for reference	None
vertical	Optional[Type['VerticalBase']]	Optional vertical class used to create this agent	None
vertical_config	Optional['VerticalConfig']	Optional vertical configuration applied	None

Raises:

Type	Description
ValueError	If orchestrator is not a valid AgentOrchestrator instance

execution_context: Any property

Return the explicit runtime execution context when available.

state: State property

Get current agent state.

Returns:

Type	Description
State	State object with stage, tool_calls_used, files_observed, etc.

vertical: Optional[Type['VerticalBase']] property

Get the vertical class used to create this agent.

Returns:

Type	Description
Optional[Type['VerticalBase']]	Vertical class or None if not created from a vertical.

vertical_config: Optional['VerticalConfig'] property

Get the vertical configuration applied to this agent.

Returns:

Type	Description
Optional['VerticalConfig']	VerticalConfig or None if not created from a vertical.

vertical_name: Optional[str] property

Get the name of the vertical used to create this agent.

Returns:

Type	Description
Optional[str]	Vertical name string or None.

event_bus: Optional['ObservabilityBus'] property

Get the ObservabilityBus for subscribing to agent events.

The ObservabilityBus provides access to all agent events including:
- Tool execution (start/end)
- State transitions
- Model requests/responses
- Errors

Returns:

Type	Description
Optional['ObservabilityBus']	ObservabilityBus instance, or None if observability is disabled

Example

def on_tool_event(event):
print(f"Tool: {event.topic} - {event.data}")

Subscribe to all tool events

agent.event_bus.backend.subscribe("tool.*", on_tool_event)

observability: Optional['ObservabilityIntegration'] property

Get the ObservabilityIntegration for advanced event handling.

Returns:

Type	Description
Optional['ObservabilityIntegration']	ObservabilityIntegration instance, or None if disabled

provider_name: Optional[str] property

Active provider name (e.g. 'zai', 'ollama') for the current session.

model: Optional[str] property

Active model name (e.g. 'glm-5.1') for the current session.

provider_base_url: Optional[str] property

Active provider base_url; non-empty when a non-default endpoint is in use.

lsp: Optional[Any] property

Get the LSP capability for code intelligence.

Returns:

Type	Description
Optional[Any]	LSPCapability instance or None

create(provider: Optional[str] = None, model: Optional[str] = None, *, temperature: float = 0.7, max_tokens: int = 4096, tools: ToolsInput = None, thinking: bool = False, airgapped: bool = False, profile: Optional[str] = None, workspace: Optional[str] = None, config: Optional[FrameworkCompatibleAgentConfig] = None, vertical: Optional[Type['VerticalBase']] = None, enable_observability: bool = True, session_id: Optional[str] = None, session_config: Optional['SessionConfig'] = None) -> 'Agent' async classmethod

Create a new Agent instance.

This is the primary way to create an Agent. For most use cases,
you only need to specify the provider.

Parameters:

Name	Type	Description	Default
provider	Optional[str]	Optional LLM provider name (anthropic, openai, ollama, google, etc.). If omitted, uses the active profile/default settings.	None
model	Optional[str]	Model identifier. If None, uses provider/profile default.	None
temperature	float	Sampling temperature (0.0 to 1.0)	0.7
max_tokens	int	Maximum tokens to generate	4096
tools	ToolsInput	Tool configuration - ToolSet, list of category names, or None	None
thinking	bool	Enable extended thinking mode (Claude only)	False
airgapped	bool	Disable network-dependent tools	False
profile	Optional[str]	Profile name from ~/.victor/profiles.yaml	None
workspace	Optional[str]	Working directory for file operations	None
config	Optional[FrameworkCompatibleAgentConfig]	Advanced configuration. Accepts AgentConfig (deprecated) or UnifiedAgentConfig (preferred). Overrides individual options.	None
vertical	Optional[Type['VerticalBase']]	Optional vertical class or name (e.g., 'coding', 'research'). When provided, the vertical's configuration (tools, system_prompt, stages) is automatically applied.	None
enable_observability	bool	Auto-initialize ObservabilityIntegration for unified event handling. Defaults to True.	True
session_id	Optional[str]	Optional session ID for event correlation.	None
session_config	Optional['SessionConfig']	Optional SessionConfig with CLI/runtime overrides. This is the PREFERRED way to pass CLI flags - immutable and traceable.	None

Returns:

Type	Description
'Agent'	Configured Agent instance

Raises:

Type	Description
ProviderError	If provider initialization fails
AgentError	If configuration is invalid

Example

Simple

agent = await Agent.create()

With provider

agent = await Agent.create(provider="openai", model="gpt-4-turbo")

With tools

agent = await Agent.create(tools=ToolSet.coding())

With SessionConfig (preferred for CLI/runtime overrides)

from victor.framework.session_config import SessionConfig
config = SessionConfig.from_cli_flags(tool_budget=50, enable_smart_routing=True)
agent = await Agent.create(session_config=config)

With vertical (domain-specific assistant)

agent = await Agent.create(vertical="coding")

from_orchestrator(orchestrator: 'AgentOrchestrator') -> 'Agent' classmethod

Create an Agent wrapper around an existing AgentOrchestrator.

This is the escape hatch for users who have already configured
an AgentOrchestrator and want to use the simplified API.

Parameters:

Name	Type	Description	Default
orchestrator	'AgentOrchestrator'	Existing AgentOrchestrator instance	required

Returns:

Type	Description
'Agent'	Agent wrapping the orchestrator

create_team(name: str, goal: str, members: List['TeamMemberSpec'], *, formation: 'TeamFormation' = None, provider: str = 'anthropic', model: Optional[str] = None, total_tool_budget: int = 100, max_iterations: int = 50, timeout_seconds: int = 600, shared_context: Optional[Dict[str, Any]] = None, config: Optional[FrameworkCompatibleAgentConfig] = None) -> 'AgentTeam' async classmethod

Create a multi-agent team.

This creates a team of agents that coordinate to achieve a shared goal.
Teams support different formation patterns for different workflows.

Parameters:

Name	Type	Description	Default
name	str	Human-readable team name	required
goal	str	Overall team objective	required
members	List['TeamMemberSpec']	List of TeamMemberSpec defining team composition	required
formation	'TeamFormation'	How agents coordinate (SEQUENTIAL, PARALLEL, HIERARCHICAL, PIPELINE)	None
provider	str	LLM provider for all team members	'anthropic'
model	Optional[str]	Model identifier for all team members	None
total_tool_budget	int	Total tool calls across all members	100
max_iterations	int	Maximum total iterations	50
timeout_seconds	int	Maximum execution time	600
shared_context	Optional[Dict[str, Any]]	Initial context shared with all members	None
config	Optional[FrameworkCompatibleAgentConfig]	Advanced configuration for the orchestrator	None

Returns:

Type	Description
'AgentTeam'	AgentTeam ready for execution

Example - Sequential research team

from victor.framework.teams import TeamMemberSpec, TeamFormation

team = await Agent.create_team(
name="Code Analysis",
goal="Analyze the authentication module",
members=[
TeamMemberSpec(role="researcher", goal="Find auth code"),
TeamMemberSpec(role="analyzer", goal="Analyze patterns"),
TeamMemberSpec(role="reviewer", goal="Summarize findings"),
],
formation=TeamFormation.SEQUENTIAL,
)
result = await team.run()

Example - Pipeline for feature implementation

team = await Agent.create_team(
name="Feature Implementation",
goal="Implement user authentication",
members=[
TeamMemberSpec(role="researcher", goal="Find auth patterns"),
TeamMemberSpec(role="planner", goal="Design implementation"),
TeamMemberSpec(role="executor", goal="Write the code"),
TeamMemberSpec(role="reviewer", goal="Review and fix"),
],
formation=TeamFormation.PIPELINE,
)

async for event in team.stream():
print(f"{event.type}: {event.message}")

Example - Hierarchical with supervisor

from victor.framework.teams import TeamAgentCategory

team = await Agent.create_team(
name="Project Team",
goal="Build a REST API",
members=[
TeamMemberSpec(
role="planner",
goal="Coordinate team",
agent_category=TeamAgentCategory.SUPERVISOR,
),
TeamMemberSpec(role="researcher", goal="Research API patterns"),
TeamMemberSpec(role="executor", goal="Implement endpoints"),
TeamMemberSpec(role="reviewer", goal="Test and review"),
],
formation=TeamFormation.HIERARCHICAL,
)

run(prompt: str, *, context: Optional[Dict[str, Any]] = None, runtime_context_overrides: Optional[Dict[str, Any]] = None) -> TaskResult async

Run a task and return the complete result.

This is the simplest way to use the agent - provide a prompt and
get back the complete response including any tool results.

Parameters:

Name	Type	Description	Default
prompt	str	What the agent should do	required
context	Optional[Dict[str, Any]]	Optional context dict (files, variables, etc.)	None
runtime_context_overrides	Optional[Dict[str, Any]]	Optional scoped runtime hints for this turn, such as named prompt overlays.	None

Returns:

Type	Description
TaskResult	TaskResult with content, tool_calls, and metadata

Example

result = await agent.run("Explain the authentication flow")
print(result.content)

With context

result = await agent.run(
"Fix the bug in this code",
context={"file": "auth.py", "error": "IndexError"}
)

stream(prompt: str, *, context: Optional[Dict[str, Any]] = None) -> AsyncIterator[AgentExecutionEvent] async

Stream events as the agent processes a task.

This provides real-time visibility into the agent's reasoning
and actions. Events include thinking, tool calls, content, and errors.

Parameters:

Name	Type	Description	Default
prompt	str	What the agent should do	required
context	Optional[Dict[str, Any]]	Optional context dict	None

Yields:

Type	Description
AsyncIterator[AgentExecutionEvent]	AgentExecutionEvent objects representing agent actions

Example

async for event in agent.stream("Analyze this codebase"):
if event.type == EventType.THINKING:
print(f"Thinking: {event.content[:50]}...")
elif event.type == EventType.TOOL_CALL:
print(f"Calling {event.tool_name}")
elif event.type == EventType.TOOL_RESULT:
print(f"Result: {event.result[:100]}...")
elif event.type == EventType.CONTENT:
print(event.content, end="", flush=True)

chat(prompt: str) -> 'ChatSession'

Start an interactive chat session.

Returns a ChatSession that maintains conversation context
across multiple turns.

Parameters:

Name	Type	Description	Default
prompt	str	Initial message	required

Returns:

Type	Description
'ChatSession'	ChatSession for multi-turn conversation

Example

session = agent.chat("Let's refactor the auth module")
response = await session.send("First, show me the current code")
response = await session.send("Now extract the validation logic")

create_session(initial_prompt: Optional[str] = None, **kwargs: Any) -> 'AgentSession'

Create the canonical multi-turn session implementation.

Parameters:

Name	Type	Description	Default
initial_prompt	Optional[str]	Optional initial prompt for the first turn.	None
**kwargs	Any	Additional AgentSession constructor options.	{}

Returns:

Type	Description
'AgentSession'	AgentSession with shared runtime/state behavior.

run_oneshot(prompt: str, *, context: Optional[Dict[str, Any]] = None) -> TaskResult async

Execute a single-turn task without maintaining conversation state.

This is a convenience method that wraps run() for one-shot tasks
where you don't need to maintain conversation context across calls.

Parameters:

Name	Type	Description	Default
prompt	str	What the agent should do	required
context	Optional[Dict[str, Any]]	Optional context dict (files, variables, etc.)	None

Returns:

Type	Description
TaskResult	TaskResult with content, tool_calls, and metadata

Example

Single API call to get a response

result = await agent.run_oneshot("Explain quantum computing")
print(result.content)

With context

result = await agent.run_oneshot(
"What's wrong with this code?",
context={"file": "auth.py", "error": "NullReferenceException"}
)

run_interactive(initial_prompt: str) -> 'ChatSession' async

Start an interactive multi-turn conversation session.

This is a convenience method that creates a ChatSession with an
initial prompt, allowing you to send multiple messages while
maintaining conversation context.

Parameters:

Name	Type	Description	Default
initial_prompt	str	The first message to start the conversation	required

Returns:

Type	Description
'ChatSession'	ChatSession for multi-turn conversation

Example

session = await agent.run_interactive("Help me refactor this code")

Continue the conversation

response1 = await session.send("What should we extract first?")
response2 = await session.send("Now apply that change")

Or stream responses

async for event in session.stream("Show me the full diff"):
if event.type == EventType.CONTENT:
print(event.content, end="")

on_state_change(callback: StateObserver) -> Callable[[], None]

Register a callback for state changes.

Parameters:

Name	Type	Description	Default
callback	StateObserver	Function called with (old_state, new_state)	required

Returns:

Type	Description
Callable[[], None]	Unsubscribe function

Example

def log_state(old, new):
print(f"State: {old.stage} -> {new.stage}")

unsubscribe = agent.on_state_change(log_state)

Later: unsubscribe()

switch_model(provider: str, model: str) -> None async

Switch to a different model.

Parameters:

Name	Type	Description	Default
provider	str	New provider name	required
model	str	New model identifier	required

set_tools(tools: Union[ToolSet, List[str]]) -> None

Update available tools.

Parameters:

Name	Type	Description	Default
tools	Union[ToolSet, List[str]]	ToolSet or list of category names	required

get_orchestrator() -> 'AgentOrchestrator'

Get the underlying AgentOrchestrator for advanced usage.

This provides access to all internal components when the
simplified API is insufficient.

Returns:

Type	Description
'AgentOrchestrator'	AgentOrchestrator instance

Example

orchestrator = agent.get_orchestrator()

Access decomposed components

controller = orchestrator.conversation_controller
pipeline = orchestrator.tool_pipeline

Access internal state

metrics = orchestrator.streaming_controller.get_session_history()

set_tool_budget(budget: int, *, user_override: bool = False) -> None

Set the maximum number of tool calls allowed.

Parameters:

Name	Type	Description	Default
budget	int	Maximum tool calls allowed	required
user_override	bool	Whether this is a user-specified override (takes precedence)	False

Example

agent = await Agent.create()
agent.set_tool_budget(50)

set_max_iterations(max_iterations: int, *, user_override: bool = False) -> None

Set the maximum number of agentic loop iterations.

Parameters:

Name	Type	Description	Default
max_iterations	int	Maximum iterations allowed	required
user_override	bool	Whether this is a user-specified override (takes precedence)	False

Example

agent = await Agent.create()
agent.set_max_iterations(20)

supports_streaming() -> bool

Check if the current provider supports streaming responses.

Returns:

Type	Description
bool	True if streaming is supported, False otherwise

Example

agent = await Agent.create()
if agent.supports_streaming():
async for event in agent.stream("Hello"):
print(event.content)

start_embedding_preload() -> None

Warm embedding-dependent runtime state when supported.

This is primarily used by chat surfaces to front-load semantic search
initialization without exposing orchestrator internals directly.

get_session_metrics() -> Dict[str, Any]

Return session-level runtime metrics when available.

set_lsp(lsp_capability: Any) -> None

Set the LSP capability for language intelligence.

Enables features like hover information, go-to-definition,
completions, and diagnostics for code operations.

Parameters:

Name	Type	Description	Default
lsp_capability	Any	LSPCapability instance	required

Example

from victor.framework.capabilities import LSPCapability

agent.set_lsp(LSPCapability())

subscribe_to_events(category: str, handler: Callable[[Any], None]) -> Optional[Callable[[], None]]

Subscribe to events of a specific category.

Convenience method for subscribing to EventBus events without
directly importing observability types.

Parameters:

Name	Type	Description	Default
category	str	Event category, wildcard alias, or topic pattern. Examples: "TOOL", "security_scan", "ALL", "tool.*"	required
handler	Callable[[Any], None]	Callback function receiving VictorEvent	required

Returns:

Type	Description
Optional[Callable[[], None]]	Unsubscribe function, or None if observability is disabled

Example

def log_tools(event):
print(f"Tool called: {event.name}")

unsubscribe = agent.subscribe_to_events("TOOL", log_tools)

Later: unsubscribe()

run_workflow(workflow_name: str, context: Optional[Dict[str, Any]] = None, *, timeout: Optional[float] = None) -> Dict[str, Any] async

Run a workflow by name from the vertical's workflow provider.

This executes a multi-step workflow defined by the vertical,
coordinating multiple agents through a DAG of operations.

Parameters:

Name	Type	Description	Default
workflow_name	str	Name of the workflow to run (e.g., "feature_implementation")	required
context	Optional[Dict[str, Any]]	Initial context data for the workflow	None
timeout	Optional[float]	Overall timeout in seconds (None = no limit)	None

Returns:

Type	Description
Dict[str, Any]	Dict with workflow result including outputs, success status, and metrics

Raises:

Type	Description
AgentError	If no vertical is configured or workflow not found

Example

Run a feature implementation workflow

result = await agent.run_workflow(
"feature_implementation",
context={"feature": "Add user authentication"}
)
print(result["success"])
print(result["outputs"])

Run an EDA workflow for data analysis

result = await agent.run_workflow(
"eda_workflow",
context={"data_file": "sales.csv"}
)

run_team(team_name: str, goal: str, *, context: Optional[Dict[str, Any]] = None, timeout_seconds: int = 600) -> Dict[str, Any] async

Run a pre-configured team from the vertical's team specs.

This creates and executes a multi-agent team defined by the vertical,
using the team's formation pattern and member specifications.

Parameters:

Name	Type	Description	Default
team_name	str	Name of the team spec (e.g., "feature_team", "bug_fix_team")	required
goal	str	Specific goal for this team execution	required
context	Optional[Dict[str, Any]]	Initial shared context for team members	None
timeout_seconds	int	Maximum execution time	600

Returns:

Type	Description
Dict[str, Any]	Dict with team result including final output and member contributions

Raises:

Type	Description
AgentError	If no vertical is configured or team not found

Example

Run a feature implementation team

result = await agent.run_team(
"feature_team",
goal="Implement user authentication with JWT",
context={"target_dir": "src/auth/"}
)

Run a bug fix team

result = await agent.run_team(
"bug_fix_team",
goal="Fix the login timeout issue",
context={"error_log": "TimeoutError at auth.py:45"}
)

get_available_workflows() -> List[str]

Get list of available workflow names from the vertical.

Returns:

Type	Description
List[str]	List of workflow names, or empty list if no vertical/workflows

Example

workflows = agent.get_available_workflows()
print(workflows) # ['feature_implementation', 'bug_fix', 'code_review']

get_available_teams() -> List[str]

Get list of available team names from the vertical.

Returns:

Type	Description
List[str]	List of team names, or empty list if no vertical/teams

Example

teams = agent.get_available_teams()
print(teams) # ['feature_team', 'bug_fix_team', 'review_team']

get_coordination_suggestion(task_type: str, complexity: str, *, mode: Optional[str] = None) -> Any

Get shared framework coordination recommendations for a task.

Parameters:

Name	Type	Description	Default
task_type	str	Classified task type	required
complexity	str	Complexity level string	required
mode	Optional[str]	Optional mode override. Defaults to the runtime's current mode.	None

Returns:

Type	Description
Any	CoordinationSuggestion with team and workflow recommendations.

get_coordination_transitions(task_type: str, complexity: Optional[str] = None, *, mode: Optional[str] = None) -> Any async

Get state-passed coordination transitions for a task.

This is the framework-facing state-passed companion to
get_coordination_suggestion(). It returns the raw coordinator result
so callers can inspect transitions, confidence, and metadata without
mutating orchestrator state directly.

reset() -> None async

Reset conversation history and state.

warm_up() -> None async

Prime the KV cache for faster first responses.

For local providers (Ollama, LMStudio) with KV prefix caching, the first
API call is always cold. This sends a minimal 1-token request to prime
the KV cache with the system prompt, making subsequent calls faster.

No-op for cloud providers or when KV optimization is disabled.

graceful_shutdown() -> Dict[str, bool] async

Perform graceful shutdown of all agent components.

Delegates to orchestrator's graceful_shutdown method if available.
Flushes analytics, stops health monitoring, and cleans up resources.
Call this before application exit for a clean shutdown.

Returns:

Type	Description
Dict[str, bool]	Dictionary with shutdown status for each component.
Dict[str, bool]	Returns empty dict if orchestrator doesn't support graceful_shutdown.

shutdown() -> None async

Shutdown the agent and clean up resources.

This is an alias for close() for compatibility with code that
expects a shutdown method.

close() -> None async

Clean up resources.

AgentBuilder

victor.framework.agent_components.AgentBuilder(container: Optional['ServiceContainer'] = None)

Builder for creating Agent instances with a fluent API.

The AgentBuilder provides a flexible, chainable interface for configuring
agents. It supports presets for common configurations and allows fine-grained
control over all agent options.

Phase 8.2: Enhanced with ServiceContainer integration for dependency injection.
When a container is provided, the builder uses container-managed services for:
- Tool configuration (ToolConfiguratorService)
- Event handling (EventRegistryService)
- Session management (AgentSessionService)

Example

Basic usage

agent = await AgentBuilder().provider("anthropic").build()

With preset

agent = await AgentBuilder().preset(BuilderPreset.CODING).build()

Fluent configuration

agent = await (
AgentBuilder()
.provider("openai")
.model("gpt-4-turbo")
.tools(["filesystem", "git"])
.thinking(True)
.build()
)

From existing options

options = AgentBuildOptions(provider="anthropic", thinking=True)
agent = await AgentBuilder.from_options(options).build()

With DI container (Phase 8.2)

from victor.framework.service_provider import configure_framework_services
container = configure_framework_services()
agent = await AgentBuilder().with_container(container).build()

Initialize builder with default options.

Parameters:

Name	Type	Description	Default
container	Optional['ServiceContainer']	Optional ServiceContainer for dependency injection. When provided, services are resolved from the container.	None

has_container: bool property

Check if a container is configured.

Returns:

Type	Description
bool	True if a ServiceContainer is set

from_options(options: AgentBuildOptions, container: Optional['ServiceContainer'] = None) -> 'AgentBuilder' classmethod

Create builder from existing options.

Parameters:

Name	Type	Description	Default
options	AgentBuildOptions	Pre-configured options	required
container	Optional['ServiceContainer']	Optional ServiceContainer for dependency injection	None

Returns:

Type	Description
'AgentBuilder'	Builder initialized with options

from_container(container: 'ServiceContainer') -> 'AgentBuilder' classmethod

Create builder from a ServiceContainer.

This is the recommended way to create builders when using DI.
The container is used to resolve all framework services.

Parameters:

Name	Type	Description	Default
container	'ServiceContainer'	ServiceContainer with framework services registered	required

Returns:

Type	Description
'AgentBuilder'	Builder with container integration

Example

container = configure_framework_services()
builder = AgentBuilder.from_container(container)
agent = await builder.provider("anthropic").build()

preset(preset: BuilderPreset) -> 'AgentBuilder'

Apply a pre-defined configuration preset.

Presets can be combined - later presets override earlier settings.

Parameters:

Name	Type	Description	Default
preset	BuilderPreset	Configuration preset to apply	required

Returns:

Type	Description
'AgentBuilder'	Self for chaining

provider(name: str) -> 'AgentBuilder'

Set the LLM provider.

Parameters:

Name	Type	Description	Default
name	str	Provider name (anthropic, openai, ollama, google, etc.)	required

Returns:

Type	Description
'AgentBuilder'	Self for chaining

model(name: str) -> 'AgentBuilder'

Set the model identifier.

Parameters:

Name	Type	Description	Default
name	str	Model identifier	required

Returns:

Type	Description
'AgentBuilder'	Self for chaining

temperature(value: float) -> 'AgentBuilder'

Set sampling temperature.

Parameters:

Name	Type	Description	Default
value	float	Temperature (0.0 to 1.0)	required

Returns:

Type	Description
'AgentBuilder'	Self for chaining

max_tokens(value: int) -> 'AgentBuilder'

Set maximum tokens to generate.

Parameters:

Name	Type	Description	Default
value	int	Maximum tokens	required

Returns:

Type	Description
'AgentBuilder'	Self for chaining

tools(tools: ToolsInput) -> 'AgentBuilder'

Set available tools.

Parameters:

Name	Type	Description	Default
tools	ToolsInput	ToolSet, list of categories, or None	required

Returns:

Type	Description
'AgentBuilder'	Self for chaining

default_tools() -> 'AgentBuilder'

Use default tool set.

Returns:

Type	Description
'AgentBuilder'	Self for chaining

minimal_tools() -> 'AgentBuilder'

Use minimal tool set.

Returns:

Type	Description
'AgentBuilder'	Self for chaining

full_tools() -> 'AgentBuilder'

Use full tool set.

Returns:

Type	Description
'AgentBuilder'	Self for chaining

airgapped_tools() -> 'AgentBuilder'

Use airgapped tool set (no network).

Returns:

Type	Description
'AgentBuilder'	Self for chaining

add_tool_filter(tool_filter: Any) -> 'AgentBuilder'

Add a tool filter for filtering tools during build.

Tool filters are applied when configuring tools on the orchestrator.
This enables runtime tool filtering based on security, cost, or custom criteria.

Parameters:

Name	Type	Description	Default
tool_filter	Any	A filter implementing ToolFilterProtocol	required

Returns:

Type	Description
'AgentBuilder'	Self for chaining

Example

from victor.framework.tool_config import AirgappedFilter, CostTierFilter
builder = (
AgentBuilder()
.add_tool_filter(AirgappedFilter())
.add_tool_filter(CostTierFilter(max_tier="MEDIUM"))
.build()
)

with_container(container: 'ServiceContainer') -> 'AgentBuilder'

Set the ServiceContainer for dependency injection.

When a container is set, the builder uses container-managed services
for tool configuration, event handling, and session management.

Parameters:

Name	Type	Description	Default
container	'ServiceContainer'	ServiceContainer with framework services registered	required

Returns:

Type	Description
'AgentBuilder'	Self for chaining

Example

from victor.framework.service_provider import configure_framework_services
container = configure_framework_services()
agent = await AgentBuilder().with_container(container).build()

thinking(enabled: bool = True) -> 'AgentBuilder'

Enable extended thinking mode.

Parameters:

Name	Type	Description	Default
enabled	bool	Whether to enable thinking	True

Returns:

Type	Description
'AgentBuilder'	Self for chaining

airgapped(enabled: bool = True) -> 'AgentBuilder'

Enable airgapped mode (no network).

Parameters:

Name	Type	Description	Default
enabled	bool	Whether to enable airgapped mode	True

Returns:

Type	Description
'AgentBuilder'	Self for chaining

profile(name: str) -> 'AgentBuilder'

Use a configuration profile.

Parameters:

Name	Type	Description	Default
name	str	Profile name from ~/.victor/profiles.yaml	required

Returns:

Type	Description
'AgentBuilder'	Self for chaining

workspace(path: str) -> 'AgentBuilder'

Set working directory for file operations.

Parameters:

Name	Type	Description	Default
path	str	Working directory path	required

Returns:

Type	Description
'AgentBuilder'	Self for chaining

config(config: FrameworkCompatibleAgentConfig) -> 'AgentBuilder'

Set advanced configuration.

Parameters:

Name	Type	Description	Default
config	FrameworkCompatibleAgentConfig	AgentConfig (deprecated) or UnifiedAgentConfig instance.	required

Returns:

Type	Description
'AgentBuilder'	Self for chaining

vertical(vertical_class: Type['VerticalBase']) -> 'AgentBuilder'

Use a domain-specific vertical.

Parameters:

Name	Type	Description	Default
vertical_class	Type['VerticalBase']	Vertical class or name string	required

Returns:

Type	Description
'AgentBuilder'	Self for chaining

system_prompt(prompt: str) -> 'AgentBuilder'

Set custom system prompt.

Parameters:

Name	Type	Description	Default
prompt	str	Custom system prompt	required

Returns:

Type	Description
'AgentBuilder'	Self for chaining

with_observability(enabled: bool = True) -> 'AgentBuilder'

Enable observability integration.

Parameters:

Name	Type	Description	Default
enabled	bool	Whether to enable	True

Returns:

Type	Description
'AgentBuilder'	Self for chaining

session_id(session_id: str) -> 'AgentBuilder'

Set session ID for event correlation.

Parameters:

Name	Type	Description	Default
session_id	str	Session identifier	required

Returns:

Type	Description
'AgentBuilder'	Self for chaining

on_enter_stage(callback: Callable[[str, Dict], None]) -> 'AgentBuilder'

Register callback for stage entry.

Parameters:

Name	Type	Description	Default
callback	Callable[[str, Dict], None]	Function called on stage entry	required

Returns:

Type	Description
'AgentBuilder'	Self for chaining

on_exit_stage(callback: Callable[[str, Dict], None]) -> 'AgentBuilder'

Register callback for stage exit.

Parameters:

Name	Type	Description	Default
callback	Callable[[str, Dict], None]	Function called on stage exit	required

Returns:

Type	Description
'AgentBuilder'	Self for chaining

on_transition(callback: Callable[[str, str, Dict], None]) -> 'AgentBuilder'

Register callback for state transitions.

Parameters:

Name	Type	Description	Default
callback	Callable[[str, str, Dict], None]	Function called on transition	required

Returns:

Type	Description
'AgentBuilder'	Self for chaining

metadata(key: str, value: Any) -> 'AgentBuilder'

Add metadata to the agent.

Parameters:

Name	Type	Description	Default
key	str	Metadata key	required
value	Any	Metadata value	required

Returns:

Type	Description
'AgentBuilder'	Self for chaining

get_options() -> AgentBuildOptions

Get current build options.

Returns:

Type	Description
AgentBuildOptions	Copy of build options

build() -> 'Agent' async

Build and return the configured Agent.

When a container is available (via with_container() or from_container()),
the build process uses container-managed services for:
- Tool configuration (ToolConfiguratorService with filters)
- Event handling (EventRegistryService for event conversion)

Returns:

Type	Description
'Agent'	Configured Agent instance

Raises:

Type	Description
AgentError	If build fails

StateGraph

victor.framework.graph.StateGraph(state_schema: Optional[Type[StateType]] = None, config_schema: Optional[Type] = None, metadata: Optional[Dict[str, Any]] = None)

Bases: Generic[StateType]

StateGraph builder for creating stateful workflows.

Provides a LangGraph-compatible API for building graph workflows
with typed state, cyclic support, and checkpointing.

Example

graph = StateGraph(AgentState)
graph.add_node("analyze", analyze_func)
graph.add_node("execute", execute_func)
graph.add_edge("analyze", "execute")
graph.add_conditional_edge(
"execute",
should_retry,
{"retry": "analyze", "done": END}
)
graph.set_entry_point("analyze")

app = graph.compile()
result = await app.invoke(initial_state)

Initialize StateGraph.

Parameters:

Name	Type	Description	Default
state_schema	Optional[Type[StateType]]	Optional type for state validation	None
config_schema	Optional[Type]	Optional type for config validation	None
metadata	Optional[Dict[str, Any]]	Optional graph-level metadata	None

node_ids: List[str] property

Return the graph's node identifiers.

add_node(node_id: str, func: Callable[[StateType], Union[StateType, Awaitable[StateType]]], **metadata: Any) -> 'StateGraph[StateType]'

Add a node to the graph.

Parameters:

Name	Type	Description	Default
node_id	str	Unique node identifier	required
func	Callable[[StateType], Union[StateType, Awaitable[StateType]]]	Node execution function	required
**metadata	Any	Additional metadata	{}

Returns:

Type	Description
'StateGraph[StateType]'	Self for chaining

Raises:

Type	Description
ValueError	If node already exists

add_edge(source: str, target: str) -> 'StateGraph[StateType]'

Add a normal edge between nodes.

Parameters:

Name	Type	Description	Default
source	str	Source node ID	required
target	str	Target node ID (or END)	required

Returns:

Type	Description
'StateGraph[StateType]'	Self for chaining

add_conditional_edge(source: str, condition: Callable[[StateType], str], branches: Dict[str, str]) -> 'StateGraph[StateType]'

Add a conditional edge with multiple branches.

Parameters:

Name	Type	Description	Default
source	str	Source node ID	required
condition	Callable[[StateType], str]	Function that returns branch name	required
branches	Dict[str, str]	Mapping from branch names to target node IDs	required

Returns:

Type	Description
'StateGraph[StateType]'	Self for chaining

add_subgraph(node_id: str, compiled_graph: 'CompiledGraph', *, input_mapper: Optional[Callable[[Dict[str, Any]], Dict[str, Any]]] = None, output_mapper: Optional[Callable[[Dict[str, Any]], Dict[str, Any]]] = None, **metadata: Any) -> 'StateGraph[StateType]'

Add a subgraph node for modular graph-of-graphs composition.

The inner compiled_graph is invoked via
:meth:CompiledGraph.invoke when this node executes.

Parameters:

Name	Type	Description	Default
node_id	str	Unique node identifier	required
compiled_graph	'CompiledGraph'	A pre-compiled graph to run as a subgraph	required
input_mapper	Optional[Callable[[Dict[str, Any]], Dict[str, Any]]]	Optional function to map parent state to subgraph input	None
output_mapper	Optional[Callable[[Dict[str, Any]], Dict[str, Any]]]	Optional function to map subgraph output back to parent	None
**metadata	Any	Additional metadata	{}

Returns:

Type	Description
'StateGraph[StateType]'	Self for chaining

Raises:

Type	Description
ValueError	If node already exists

add_state_merger(func: Callable[[Dict[str, Any], List[Dict[str, Any]]], Dict[str, Any]]) -> 'StateGraph[StateType]'

Set a custom state merger for fan-out / parallel branches.

The merger receives the base state and a list of branch result
states and must return the merged state.

Parameters:

Name	Type	Description	Default
func	Callable[[Dict[str, Any], List[Dict[str, Any]]], Dict[str, Any]]	Merger function (base_state, branch_states) -> merged	required

Returns:

Type	Description
'StateGraph[StateType]'	Self for chaining

set_entry_point(node_id: str) -> 'StateGraph[StateType]'

Set the entry point node.

Parameters:

Name	Type	Description	Default
node_id	str	Node to start execution from	required

Returns:

Type	Description
'StateGraph[StateType]'	Self for chaining

set_finish_point(node_id: str) -> 'StateGraph[StateType]'

Set a node as finish point (adds edge to END).

Parameters:

Name	Type	Description	Default
node_id	str	Node that finishes the graph	required

Returns:

Type	Description
'StateGraph[StateType]'	Self for chaining

compile(checkpointer: Optional[CheckpointerProtocol] = None, strict_edges: bool = False, **config_kwargs: Any) -> CompiledGraph[StateType]

Compile the graph for execution.

Validates graph structure and creates optimized execution plan.

Parameters:

Name	Type	Description	Default
checkpointer	Optional[CheckpointerProtocol]	Optional checkpointer for persistence	None
strict_edges	bool	If True, raise EdgeResolutionError when a conditional edge doesn't match any case instead of falling through to END.	False
**config_kwargs	Any	Additional config options	{}

Returns:

Type	Description
CompiledGraph[StateType]	CompiledGraph ready for execution

Raises:

Type	Description
ValueError	If graph is invalid

from_schema(schema: Union[Dict[str, Any], str], state_schema: Optional[Type[StateType]] = None, node_registry: Optional[Dict[str, Callable]] = None, condition_registry: Optional[Dict[str, Callable]] = None) -> 'StateGraph[StateType]' classmethod

Create StateGraph from schema dictionary or YAML string.

This enables dynamic graph generation from serialized schemas,
supporting Phase 3.0 requirements for workflow persistence and
external graph definition.

Parameters:

Name	Type	Description	Default
schema	Union[Dict[str, Any], str]	Either a dictionary schema or YAML string containing: - nodes: List of node definitions with id and type - edges: List of edge definitions with source, target, type - entry_point: Starting node ID - Optional: state_schema, metadata	required
state_schema	Optional[Type[StateType]]	Optional TypedDict type for state validation	None
node_registry	Optional[Dict[str, Callable]]	Registry of node functions (for 'function' type nodes) Maps node function names to callable functions	None
condition_registry	Optional[Dict[str, Callable]]	Registry of condition functions (for conditional edges) Maps condition function names to callable functions	None

Returns:

Type	Description
'StateGraph[StateType]'	StateGraph instance ready for compilation

Raises:

Type	Description
ValueError	If schema is invalid or missing required fields
TypeError	If node/condition types are unsupported

Example

Define schema

schema = {
"nodes": [
{"id": "analyze", "type": "function", "func": "analyze_task"},
{"id": "execute", "type": "function", "func": "execute_task"},
],
"edges": [
{"source": "analyze", "target": "execute", "type": "normal"},
{
"source": "execute",
"target": {"retry": "analyze", "done": "end"},
"type": "conditional",
"condition": "should_retry"
}
],
"entry_point": "analyze"
}

Create registries

node_registry = {
"analyze_task": analyze_task_func,
"execute_task": execute_task_func,
}
condition_registry = {
"should_retry": should_retry_func,
}

Deserialize

graph = StateGraph.from_schema(
schema,
state_schema=AgentState,
node_registry=node_registry,
condition_registry=condition_registry
)

Compile and execute

app = graph.compile()
result = await app.invoke(initial_state)

Example with YAML

yaml_schema = """
nodes:
- id: analyze
type: function
func: analyze_task
- id: execute
type: function
func: execute_task
edges:
- source: analyze
target: execute
type: normal
- source: execute
target:
retry: analyze
done: end
type: conditional
condition: should_retry
entry_point: analyze
"""

graph = StateGraph.from_schema(
yaml_schema,
node_registry=node_registry,
condition_registry=condition_registry
)

WorkflowEngine

victor.framework.workflow_engine.WorkflowEngine(config: Optional[WorkflowEngineConfig] = None, hitl_handler: Optional['HITLHandler'] = None, cache_manager: Optional['WorkflowCacheManager'] = None, runner_registry: Optional['NodeRunnerRegistry'] = None)

High-level facade for workflow execution.

Provides a unified API for executing workflows defined as:
- YAML files (declarative)
- StateGraph objects (programmatic)
- WorkflowDefinition objects (legacy)

Features:
- Automatic HITL integration for approval nodes
- Result caching for repeated executions
- Checkpointing for recovery
- Event streaming for real-time updates
- Parallel node execution

Initialize WorkflowEngine.

Parameters:

Name	Type	Description	Default
config	Optional[WorkflowEngineConfig]	Engine configuration.	None
hitl_handler	Optional['HITLHandler']	Custom HITL handler for approval nodes.	None
cache_manager	Optional['WorkflowCacheManager']	Custom cache manager for results.	None
runner_registry	Optional['NodeRunnerRegistry']	Optional NodeRunner registry for unified execution.	None

config: WorkflowEngineConfig property

Get the engine configuration.

execute_yaml(yaml_path: Union[str, Path], initial_state: Optional[Dict[str, Any]] = None, workflow_name: Optional[str] = None, condition_registry: Optional[Dict[str, Callable]] = None, transform_registry: Optional[Dict[str, Callable]] = None, use_unified_compiler: bool = True, **kwargs: Any) -> WorkflowExecutionResult async

Execute a workflow from YAML file.

Uses UnifiedWorkflowCompiler for consistent compilation and caching,
then executes via CompiledGraph.invoke().

Parameters:

Name	Type	Description	Default
yaml_path	Union[str, Path]	Path to YAML workflow file.	required
initial_state	Optional[Dict[str, Any]]	Initial workflow state.	None
workflow_name	Optional[str]	Specific workflow to load from file.	None
condition_registry	Optional[Dict[str, Callable]]	Custom condition functions.	None
transform_registry	Optional[Dict[str, Callable]]	Custom transform functions.	None
use_unified_compiler	bool	Whether to use unified compiler (default True). Set to False for backward compatibility with coordinator.	True
**kwargs	Any	Additional execution parameters.	{}

Returns:

Type	Description
WorkflowExecutionResult	WorkflowExecutionResult with final state and metadata.

execute_graph(graph: 'CompiledGraph', initial_state: Optional[Dict[str, Any]] = None, **kwargs: Any) -> WorkflowExecutionResult async

Execute a compiled StateGraph.

Delegates to GraphTurnExecutor for SRP-compliant execution
with LSP-compliant polymorphic result handling.

Parameters:

Name	Type	Description	Default
graph	'CompiledGraph'	Compiled StateGraph to execute.	required
initial_state	Optional[Dict[str, Any]]	Initial workflow state.	None
**kwargs	Any	Additional execution parameters.	{}

Returns:

Type	Description
WorkflowExecutionResult	WorkflowExecutionResult with final state and metadata.

execute_definition(workflow: 'WorkflowDefinition', initial_state: Optional[Dict[str, Any]] = None, **kwargs: Any) -> WorkflowExecutionResult async

Execute a WorkflowDefinition.

Parameters:

Name	Type	Description	Default
workflow	'WorkflowDefinition'	WorkflowDefinition to execute.	required
initial_state	Optional[Dict[str, Any]]	Initial workflow state.	None
**kwargs	Any	Additional execution parameters.	{}

Returns:

Type	Description
WorkflowExecutionResult	WorkflowExecutionResult with final state and metadata.

execute_workflow_graph(graph: 'WorkflowGraph', initial_state: Optional[Dict[str, Any]] = None, use_node_runners: bool = False, **kwargs: Any) -> WorkflowExecutionResult async

Execute a WorkflowGraph via CompiledGraph (unified execution path).

Delegates to GraphTurnExecutor for SRP-compliant execution.
This method compiles a WorkflowGraph to CompiledGraph and executes
it through the single CompiledGraph.invoke() engine, providing a
unified execution path for all workflow types.

Parameters:

Name	Type	Description	Default
graph	'WorkflowGraph'	WorkflowGraph to compile and execute.	required
initial_state	Optional[Dict[str, Any]]	Initial workflow state.	None
use_node_runners	bool	Whether to use NodeRunner protocol for execution.	False
**kwargs	Any	Additional execution parameters.	{}

Returns:

Type	Description
WorkflowExecutionResult	WorkflowExecutionResult with final state and metadata.

Example

from victor.workflows.graph_dsl import WorkflowGraph, State

@dataclass
class MyState(State):
value: int = 0

graph = WorkflowGraph(MyState)
graph.add_node("process", lambda s: s)
graph.set_entry_point("process")
graph.set_finish_point("process")

result = await engine.execute_workflow_graph(graph, {"value": 42})

execute_definition_compiled(workflow: 'WorkflowDefinition', initial_state: Optional[Dict[str, Any]] = None, **kwargs: Any) -> WorkflowExecutionResult async

Execute a WorkflowDefinition via CompiledGraph (unified execution path).

Delegates to GraphTurnExecutor for SRP-compliant execution.
This method compiles a WorkflowDefinition to CompiledGraph and executes
it through the single CompiledGraph.invoke() engine.

Parameters:

Name	Type	Description	Default
workflow	'WorkflowDefinition'	WorkflowDefinition to compile and execute.	required
initial_state	Optional[Dict[str, Any]]	Initial workflow state.	None
**kwargs	Any	Additional execution parameters.	{}

Returns:

Type	Description
WorkflowExecutionResult	WorkflowExecutionResult with final state and metadata.

set_runner_registry(registry: 'NodeRunnerRegistry') -> None

Set the NodeRunner registry for unified execution.

Parameters:

Name	Type	Description	Default
registry	'NodeRunnerRegistry'	NodeRunnerRegistry with configured runners.	required

stream_yaml(yaml_path: Union[str, Path], initial_state: Optional[Dict[str, Any]] = None, workflow_name: Optional[str] = None, condition_registry: Optional[Dict[str, Callable]] = None, transform_registry: Optional[Dict[str, Callable]] = None, use_unified_compiler: bool = True, **kwargs: Any) -> AsyncIterator[WorkflowEvent] async

Stream events from YAML workflow execution.

Uses UnifiedWorkflowCompiler for consistent compilation and caching,
then streams via CompiledGraph.stream().

Parameters:

Name	Type	Description	Default
yaml_path	Union[str, Path]	Path to YAML workflow file.	required
initial_state	Optional[Dict[str, Any]]	Initial workflow state.	None
workflow_name	Optional[str]	Specific workflow to load from file.	None
condition_registry	Optional[Dict[str, Callable]]	Custom condition functions.	None
transform_registry	Optional[Dict[str, Callable]]	Custom transform functions.	None
use_unified_compiler	bool	Whether to use unified compiler (default True). Set to False for backward compatibility with coordinator.	True
**kwargs	Any	Additional execution parameters.	{}

Yields:

Type	Description
AsyncIterator[WorkflowEvent]	WorkflowEvent for each execution step.

stream_graph(graph: 'CompiledGraph', initial_state: Optional[Dict[str, Any]] = None, **kwargs: Any) -> AsyncIterator[WorkflowEvent] async

Stream events from StateGraph execution.

Delegates to GraphTurnExecutor for SRP-compliant streaming.

Parameters:

Name	Type	Description	Default
graph	'CompiledGraph'	Compiled StateGraph to execute.	required
initial_state	Optional[Dict[str, Any]]	Initial workflow state.	None
**kwargs	Any	Additional execution parameters.	{}

Yields:

Type	Description
AsyncIterator[WorkflowEvent]	WorkflowEvent for each execution step.

set_hitl_handler(handler: 'HITLHandler') -> None

Set custom HITL handler.

Parameters:

Name	Type	Description	Default
handler	'HITLHandler'	HITLHandler for approval nodes.	required

execute_with_hitl(yaml_path: Union[str, Path], initial_state: Optional[Dict[str, Any]] = None, approval_callback: Optional[Callable[[Dict[str, Any]], bool]] = None, **kwargs: Any) -> WorkflowExecutionResult async

Execute workflow with HITL approval nodes.

Delegates to HITLCoordinator for SRP-compliant execution.

Parameters:

Name	Type	Description	Default
yaml_path	Union[str, Path]	Path to YAML workflow file.	required
initial_state	Optional[Dict[str, Any]]	Initial workflow state.	None
approval_callback	Optional[Callable[[Dict[str, Any]], bool]]	Callback for approval decisions.	None
**kwargs	Any	Additional execution parameters.	{}

Returns:

Type	Description
WorkflowExecutionResult	WorkflowExecutionResult with HITL request history.

enable_caching(ttl_seconds: int = 3600) -> None

Enable result caching.

Delegates to CacheCoordinator for SRP-compliant cache management.

Parameters:

Name	Type	Description	Default
ttl_seconds	int	Cache time-to-live.	3600

disable_caching() -> None

Disable result caching.

Delegates to CacheCoordinator for SRP-compliant cache management.

clear_cache() -> None

Clear all cached results.

Delegates to CacheCoordinator for SRP-compliant cache management.

clear_workflow_cache() -> int

Clear all workflow caches via unified compiler.

Clears both definition cache (parsed YAML workflows) and
execution cache (workflow results).

Returns:

Type	Description
int	Total number of cache entries cleared.

get_workflow_cache_stats() -> Dict[str, Any]

Get workflow cache statistics.

Returns comprehensive cache statistics including:
- definition_cache: Stats for parsed workflow definitions
- execution_cache: Stats for workflow execution results
- caching_enabled: Whether caching is currently enabled

Returns:

Type	Description
Dict[str, Any]	Dictionary with cache statistics.

invalidate_yaml_cache(yaml_path: Union[str, Path]) -> int

Invalidate cached definitions for a specific YAML file.

Use this when a YAML file has been modified and the cache
should be refreshed.

Parameters:

Name	Type	Description	Default
yaml_path	Union[str, Path]	Path to YAML file to invalidate.	required

Returns:

Type	Description
int	Number of cache entries invalidated.

UI entry point

The only supported entry point for surface layers (CLI, TUI, web chat).

VictorClient

victor.framework.client.VictorClient(config: 'SessionConfig', *, container: Optional['ServiceContainer'] = None)

Unified client facade — the ONLY interface the UI layer uses.

This class makes the CLI a first-class client of the framework by:
1. Accepting SessionConfig for CLI/runtime overrides (not settings mutation)
2. Creating Agent via Agent.create() with session_config
3. Accessing services through ServiceAccessor (not orchestrator bypass)
4. Providing chat(), stream(), create_session(), run_workflow()

Architectural Guarantees

• NEVER accesses orchestrator directly
• ALWAYS uses services (ChatService, ToolService, etc.)
• ENFORCES proper service boundaries

Example

from victor.framework.client import VictorClient
from victor.framework.session_config import SessionConfig

config = SessionConfig.from_cli_flags(tool_budget=50)
async with VictorClient(config) as client:
result = await client.chat("Write a hello world")
print(result.content)

Initialize client with session config and optional DI container.

Parameters:

Name	Type	Description	Default
config	'SessionConfig'	SessionConfig with CLI/runtime overrides (immutable)	required
container	Optional['ServiceContainer']	Optional pre-built DI container. If None, one is bootstrapped lazily on first use.	None

provider_name: Optional[str] property

Active provider name for the current session (e.g. 'zai', 'ollama').

model: Optional[str] property

Active model name for the current session (e.g. 'glm-5.1').

provider_base_url: Optional[str] property

Active provider base_url (non-empty when a non-default endpoint is used).

initialize() -> 'Agent' async

Public initialization hook for UI surfaces.

set_approval_handler(handler: Any) -> None

Register a policy ASK approval handler for this session.

The handler (an async elicitation callable, e.g. a Chainlit Approve/Reject prompt)
is stored on the client and registered into the session's DI container during
initialization — after the container is bootstrapped and before the agent builds
its policy middleware — so ASK-gated tools resolve interactively instead of falling
back to ask_fallback. Must be called before the first chat()/stream().

Idempotent: the latest handler set before initialization wins. Safe to call on
every UI on_chat_start (including reconnects), since registration is deferred
to _ensure_initialized.

start_embedding_preload() -> None async

Warm embedding-dependent runtime state when supported.

get_session_metrics() -> Dict[str, Any] async

Return session-level runtime metrics when available.

get_last_turn_cost() -> Dict[str, Any]

Return the most recent per-turn cost/latency record (the C0 TaskExecutionReport).

Surfaces the canonical per-turn record — tokens, cost, duration, request count, cache
hit rate — so UI surfaces can render a cost/latency footer without reaching into the
orchestrator directly (UI-layer mandate). Returns {} when no turn has completed or
the record is unavailable.

chat(message: str, *, stream: bool = False) -> TaskResult async

Send a single message and get a response.

Uses the shared framework message-execution surface so UI callers get
the same service-first runtime resolution and output normalization as Agent.

Parameters:

Name	Type	Description	Default
message	str	User's message	required
stream	bool	If True, use streaming internally but return final result	False

Returns:

Type	Description
TaskResult	TaskResult with response content, tool calls, and metadata

stream(message: str) -> AsyncIterator[_StreamEvent] async

Send a message and yield streaming events.

Uses the shared framework runtime-event iterator so chat services and
Agent wrappers present the same event contract to UI callers.

Parameters:

Name	Type	Description	Default
message	str	User's message	required

Yields:

Type	Description
AsyncIterator[_StreamEvent]	StreamEvent instances (content, thinking, tool_call, etc.)

stream_chat(message: str) -> AsyncIterator[_RenderChunk] async

Yield renderer-compatible chunks for legacy UI streaming helpers.

create_session(initial_prompt: Optional[str] = None) -> 'AgentSession' async

Create an interactive multi-turn chat session.

Delegates to the canonical framework AgentSession so chat/session
behavior stays on one framework-owned path.

Returns:

Type	Description
'AgentSession'	AgentSession for multi-turn conversation

run_workflow(workflow_name: str, inputs: Optional[Dict[str, Any]] = None) -> Dict[str, Any] async

Run a named workflow via the Agent's workflow API.

Parameters:

Name	Type	Description	Default
workflow_name	str	Registered workflow name	required
inputs	Optional[Dict[str, Any]]	Workflow input parameters	None

Returns:

Type	Description
Dict[str, Any]	Workflow execution result

get_available_workflows() -> List[str]

List available workflow names.

get_available_verticals() -> List[str]

List available vertical names.

get_available_providers() -> List[str]

List available provider names.

reset_conversation() -> None async

Reset conversation history and state.

Uses ChatService to clear conversation context while preserving
system prompts and session configuration.

Raises:

Type	Description
RuntimeError	If client is not initialized

get_messages(limit: Optional[int] = None, role: Optional[str] = None) -> List[Any] async

Get conversation messages.

Parameters:

Name	Type	Description	Default
limit	Optional[int]	Maximum number of messages to return (most recent first)	None
role	Optional[str]	Optional filter by message role (e.g., "user", "assistant")	None

Returns:

Type	Description
List[Any]	List of message objects (type depends on Message implementation)

Raises:

Type	Description
RuntimeError	If client is not initialized

list_recent_sessions(limit: int = 10) -> List[Dict[str, Any]]

List recent stored sessions for a resume picker (cross-visit).

Reads the canonical ConversationStore directly — a read-only lookup
that does not require an initialized agent, so a surface can render
the picker before a client is bound to a session.

Parameters:

Name	Type	Description	Default
limit	int	Maximum number of sessions to return (most recent first).	10

Returns:

Type	Description
List[Dict[str, Any]]	Session summary dicts (session_id, title, message_count,
List[Dict[str, Any]]	last_activity, …); empty on any lookup failure.

resume_session(session_id: str) -> Optional[Dict[str, Any]] async

Resume a stored session so the agent recalls its prior turns.

Hydrates the live conversation from the ConversationStore via
ChatService.resume_session (which repopulates both message stores),
then stamps the resumed session_id (and conversation state, when
present) onto the orchestrator so continuation writes append to the
same session.

Parameters:

Name	Type	Description	Default
session_id	str	The stored session to resume.	required

Returns:

Type	Description
Optional[Dict[str, Any]]	The session metadata (title, message_count, …) on success, or
Optional[Dict[str, Any]]	None when the session is not found.

Raises:

Type	Description
RuntimeError	If the client is not initialized.

resume(run_id: str, decision: 'ApprovalDecision') -> 'TaskResult' async

Resume a durably-paused turn with a human approval decision (FEP-0029).

When a turn paused on a policy ASK (chat/stream returned a TaskResult with
status="awaiting_approval" + a run_id), this replays the exact persisted gated
tool call — it does not re-sample the model — and continues the turn:

• decision.approved → executes the gated call (bypassing the ASK, since the human
  approved), appends its result, and drives the model to a final answer.
• not approved → skips the call with a tool-error result and continues.

Single-use: a second resume on the same run_id raises. Returns the continued turn's
:class:TaskResult.

Parameters:

Name	Type	Description	Default
run_id	str	The resume token from the paused TaskResult.	required
decision	'ApprovalDecision'	The human's :class:ApprovalDecision.	required

Raises:

Type	Description
RuntimeError	If the client is not initialized.
ValueError	If run_id is unknown or already resumed.

close() -> None async

Clean up agent and container resources.

Waits (bounded) for any in-flight stream() to finish before tearing down
the agent/provider, so a close() racing an active stream (e.g. a UI on_chat_end
on a mid-run WebSocket disconnect) does not close the provider out from under it.

SessionConfig

victor.framework.session_config.SessionConfig(agent_profile: Optional[str] = None, tool_budget: Optional[int] = None, max_iterations: Optional[int] = None, planning_enabled: Optional[bool] = None, planning_model: Optional[str] = None, mode: Optional[str] = None, show_reasoning: bool = False, observability_logging: Optional[bool] = None, auto_skill_enabled: Optional[bool] = None, one_shot_mode: Optional[bool] = None, headless_mode: bool = False, verify_mode: str = 'none', lsp_feedback: str = 'errors', lsp_perception: bool = False, compaction: CompactionConfig = CompactionConfig(), smart_routing: SmartRoutingConfig = SmartRoutingConfig(), tool_output: ToolOutputConfig = ToolOutputConfig(), provider_override: ProviderOverrideConfig = ProviderOverrideConfig(), bayesian: BayesianConfig = BayesianConfig(), tool_approval: ToolApprovalConfig = ToolApprovalConfig(), shell_safety: ShellSafetyConfig = ShellSafetyConfig()) dataclass

Immutable capture of all CLI/runtime session overrides.

This is the single config object that the CLI (or any client)
produces and passes to Agent.create(). The framework reads it
but never mutates it — eliminating scattered settings.xxx = yyy
mutations throughout the codebase.

Attributes:

Name	Type	Description
agent_profile	Optional[str]	Agent profile name from ~/.victor/profiles.yaml (e.g., 'zai-coding', 'default').
tool_budget	Optional[int]	Override tool call budget for this session.
max_iterations	Optional[int]	Override maximum iterations for this session.
compaction	CompactionConfig	Compaction threshold overrides.
smart_routing	SmartRoutingConfig	Smart provider routing overrides.
tool_output	ToolOutputConfig	Tool output preview/pruning overrides.
planning_enabled	Optional[bool]	Enable structured planning for complex tasks.
planning_model	Optional[str]	Override model for planning tasks.
mode	Optional[str]	Initial agent mode ('build', 'plan', 'explore').
show_reasoning	bool	Show LLM reasoning/thinking content.
provider_override	ProviderOverrideConfig	Explicit provider/model/endpoint override state.
tool_preview	ProviderOverrideConfig	Shorthand to disable tool output preview.
enable_pruning	ProviderOverrideConfig	Shorthand to enable broader tool output pruning.
enable_smart_routing	ProviderOverrideConfig	Shorthand to enable smart routing.
routing_profile	ProviderOverrideConfig	Shorthand for routing profile.
fallback_chain	ProviderOverrideConfig	Shorthand for fallback provider chain.
compaction_threshold	ProviderOverrideConfig	Shorthand for compaction threshold.
adaptive_threshold	ProviderOverrideConfig	Shorthand for adaptive compaction toggle.
compaction_min_threshold	ProviderOverrideConfig	Shorthand for adaptive min threshold.
compaction_max_threshold	ProviderOverrideConfig	Shorthand for adaptive max threshold.

from_cli_flags(*, agent_profile: Optional[str] = None, tool_budget: Optional[int] = None, max_iterations: Optional[int] = None, compaction_threshold: Optional[float] = None, adaptive_threshold: Optional[bool] = None, compaction_min_threshold: Optional[float] = None, compaction_max_threshold: Optional[float] = None, enable_smart_routing: bool = False, routing_profile: str = 'balanced', fallback_chain: Optional[str] = None, tool_preview: bool = True, enable_pruning: bool = False, planning_enabled: Optional[bool] = None, planning_model: Optional[str] = None, mode: Optional[str] = None, show_reasoning: bool = False, observability_logging: Optional[bool] = None, auto_skill_enabled: Optional[bool] = None, one_shot_mode: Optional[bool] = None, headless_mode: bool = False, verify_mode: str = 'none', lsp_feedback: str = 'errors', lsp_perception: bool = False, provider: Optional[str] = None, model: Optional[str] = None, endpoint: Optional[str] = None, auth_mode: Optional[str] = None, provider_timeout: Optional[int] = None, coding_plan: bool = False, enable_bayesian: bool = True, force_bayesian: bool = False, simple_threshold: float = 0.3, complex_threshold: float = 0.7, enable_voi: bool = True, enable_correlation: bool = True, min_agents_for_bayesian: int = 2, tool_approval_enabled: bool = False, ask_on_tools: Optional[List[str]] = None, ask_fallback: str = 'deny', durable_approval: bool = False, shell_safety_profile: Optional[str] = None, shell_workspace_root: Optional[str] = None, shell_allow_network: Optional[bool] = None) -> 'SessionConfig' classmethod

Create a SessionConfig from flat CLI flags.

This is the primary factory for CLI code — collect all Typer
options and pass them here to get an immutable config object.

Parameters:

Name	Type	Description	Default
agent_profile	Optional[str]	Agent profile name from ~/.victor/profiles.yaml.	None
tool_budget	Optional[int]	Override tool call budget.	None
max_iterations	Optional[int]	Override max iterations.	None
compaction_threshold	Optional[float]	Compaction threshold (0.1-0.95).	None
adaptive_threshold	Optional[bool]	Enable adaptive compaction.	None
compaction_min_threshold	Optional[float]	Adaptive min threshold.	None
compaction_max_threshold	Optional[float]	Adaptive max threshold.	None
enable_smart_routing	bool	Enable smart routing.	False
routing_profile	str	Routing profile name.	'balanced'
fallback_chain	Optional[str]	Fallback provider chain.	None
tool_preview	bool	Show tool output previews.	True
enable_pruning	bool	Enable broader tool output pruning.	False
planning_enabled	Optional[bool]	Enable structured planning.	None
planning_model	Optional[str]	Override model for planning.	None
mode	Optional[str]	Agent mode (build/plan/explore).	None
show_reasoning	bool	Show LLM reasoning.	False
observability_logging	Optional[bool]	Enable event/observability logging for this session.	None
auto_skill_enabled	Optional[bool]	Override skill auto-selection for this session.	None
one_shot_mode	Optional[bool]	Override headless one-shot execution mode.	None
provider	Optional[str]	Override provider for this session.	None
model	Optional[str]	Override model for this session.	None
endpoint	Optional[str]	Override endpoint for local providers.	None
auth_mode	Optional[str]	Override provider auth mode.	None
provider_timeout	Optional[int]	Override provider request timeout in seconds.	None
coding_plan	bool	Enable provider-specific coding-plan endpoint mode.	False

Returns:

Type	Description
'SessionConfig'	Immutable SessionConfig instance.

config = SessionConfig.from_cli_flags(
    tool_budget=50,
    enable_smart_routing=True,
    tool_preview=False,
)
agent = await Agent.create(session_config=config)

apply_to_settings(settings: object) -> None

Apply session overrides to a Settings object.

This is the only place where Settings mutation should happen
from session config. All CLI code should call this method instead
of directly mutating settings.xxx = yyy.

Parameters:

Name	Type	Description	Default
settings	object	Application Settings instance.	required

Extension base classes

Inherit from these to add providers and tools.

BaseProvider

victor.providers.base.BaseProvider(api_key: Optional[str] = None, base_url: Optional[str] = None, timeout: int = 60, max_retries: int = 3, use_circuit_breaker: bool = True, circuit_breaker_failure_threshold: int = 5, circuit_breaker_recovery_timeout: float = 30.0, **kwargs: Any)

Bases: ABC

Abstract base class for all LLM providers.

Initialize provider.

Parameters:

Name	Type	Description	Default
api_key	Optional[str]	API key for authentication (if required)	None
base_url	Optional[str]	Base URL for API endpoints	None
timeout	int	Request timeout in seconds	60
max_retries	int	Maximum number of retry attempts	3
use_circuit_breaker	bool	Whether to enable circuit breaker protection	True
circuit_breaker_failure_threshold	int	Failures before opening circuit	5
circuit_breaker_recovery_timeout	float	Seconds before testing recovery	30.0
**kwargs	Any	Additional provider-specific options	{}

circuit_breaker: Optional[CircuitBreaker] property

Get the circuit breaker for this provider.

DEFAULT_CONTEXT_WINDOW: int = 8192 class-attribute instance-attribute

Conservative default context window when model is unknown.

Triggers semantic_select_capped strategy in the tool broadcaster, which
is safe for any model. Override per-provider with a model lookup table.

name: str abstractmethod property

Provider name.

supports_tools() -> bool

Check if provider supports tool/function calling.

Default implementation returns False. Providers that support tool calling
should override this method to return True. This follows the Interface
Segregation Principle - providers don't need to implement tool calling
if they don't support it.

Returns:

Type	Description
bool	True if provider supports tools, False otherwise (default)

supports_streaming() -> bool

Check if provider supports streaming responses.

Default implementation returns False. Providers that support streaming
should override this method to return True. This follows the Interface
Segregation Principle - providers don't need to implement streaming
if they don't support it.

Returns:

Type	Description
bool	True if provider supports streaming, False otherwise (default)

supports_vision() -> bool

Check if provider supports multimodal (image) input.

Default implementation returns False. Providers that accept messages
with images populated should override this to return True. This
follows the Interface Segregation Principle - providers don't need to
implement vision if they don't support it - and completes the base
capability contract so provider.supports_vision() is answerable on
every provider (Liskov), matching supports_tools/supports_streaming.

Returns:

Type	Description
bool	True if provider supports image input, False otherwise (default)

supports_prompt_caching() -> bool

Check if provider supports API-level prompt prefix caching.

API-level prompt caching means the provider offers a billing discount
(50-90%) on cached input tokens. For these providers, sending the full
tool set (48 tools) every call is optimal because cached tokens are
nearly free after the first call.

This is distinct from KV prefix caching (see supports_kv_prefix_caching),
which is a latency optimization at the inference engine level.

Providers WITHOUT API-level caching (Ollama, LMStudio, llama.cpp, MLX,
vLLM) should use per-turn semantic tool selection (5-12 tools) to
minimize token overhead.

Default implementation returns False.

Returns:

Type	Description
bool	True if provider has API-level cached token discounts

supports_kv_prefix_caching() -> bool

Check if provider supports KV prefix caching for latency savings.

KV prefix caching means the inference engine reuses computed key-value
state when consecutive requests share the same prompt prefix. This
reduces time-to-first-token (TTFT) but does NOT reduce cost — every
token still incurs compute on the first call.

When True, the framework should keep the system prompt + tool definitions
stable across turns within a session so the KV cache can be reused.

Both local engines (Ollama, vLLM, llama.cpp) and cloud APIs (Anthropic,
OpenAI) support this. Default is False.

Returns:

Type	Description
bool	True if provider benefits from stable prompt prefixes

cache_cost_model() -> CacheCostModel

Characterized economics for API-level prompt caching (FEP-0011).

Default derives from :meth:supports_prompt_caching so this is purely
additive and existing providers are unaffected. Override to advertise
real numbers (discount %, TTL, prefix granularity, …) so the prompt
assembler can optimize against them instead of a single boolean.

Returns:

Type	Description
CacheCostModel	A frozen :class:CacheCostModel describing this provider's
CacheCostModel	API-level prompt-caching economics.

kv_cache_cost_model() -> CacheCostModel

Characterized economics for KV prefix caching (FEP-0011).

Default derives from :meth:supports_kv_prefix_caching. Override to
advertise real numbers (KV cache is latency-only; read_discount is
typically 0.0 since there is no billing discount).

Returns:

Type	Description
CacheCostModel	A frozen :class:CacheCostModel describing this provider's
CacheCostModel	KV-prefix-caching economics.

supports_reasoning_effort(model: Optional[str] = None) -> bool

Check if a model accepts the reasoning_effort request parameter.

reasoning_effort ("low" / "medium" / "high") controls how
much a reasoning model deliberates before answering. It is only valid for
reasoning models on providers that expose it (e.g. OpenAI o-series /
GPT-5); sending it to a model that doesn't support it is an API error.

The framework consults this before forwarding a per-member
reasoning_effort so the parameter is never sent to a model that would
reject it. Default is False — providers/models opt in by overriding.

Parameters:

Name	Type	Description	Default
model	Optional[str]	Model identifier to check (provider may key support on it).	None

Returns:

Type	Description
bool	True if reasoning_effort may be forwarded for model.

get_tool_output_format() -> Any

Get preferred tool output format for this provider.

This method enables provider-specific customization of tool output
formatting, following the Strategy pattern. Default implementation
returns plain JSON format (token-efficient for cloud providers).

Providers can override to specify XML, TOON, or custom formats:
- Cloud providers (OpenAI, xAI, etc.): Use default plain JSON
- Local providers (Ollama, vLLM, llama.cpp): Override to XML format
- Experimental: Override to TOON for structured data

Returns:

Type	Description
Any	ToolOutputFormat specification (from victor.agent.format_strategies)

Example

from victor.agent.format_strategies import ToolOutputFormat, XML_FORMAT

class OllamaProvider(BaseProvider):
def get_tool_output_format(self):
# Local models trained on XML format
return XML_FORMAT

get_circuit_breaker_stats() -> Optional[Dict[str, Any]]

Get circuit breaker statistics for monitoring.

classify_error(error: Exception) -> ProviderError

Classify a raw exception into the appropriate ProviderError subtype.

Providers can override this for provider-specific error handling.
The base implementation uses a three-tier strategy:
1. Pass through existing ProviderError subtypes unchanged
2. Check HTTP status codes (if available on the exception)
3. String-based pattern matching as final fallback

Parameters:

Name	Type	Description	Default
error	Exception	The raw exception from the provider API call.	required

Returns:

Type	Description
ProviderError	A ProviderError (or subtype) wrapping the original exception.

chat(messages: List[Message], *, model: str, temperature: float = 0.7, max_tokens: int = 4096, tools: Optional[List[ToolDefinition]] = None, **kwargs: Any) -> CompletionResponse abstractmethod async

Send a chat completion request.

Parameters:

Name	Type	Description	Default
messages	List[Message]	List of conversation messages	required
model	str	Model identifier	required
temperature	float	Sampling temperature (0-2)	0.7
max_tokens	int	Maximum tokens to generate	4096
tools	Optional[List[ToolDefinition]]	Available tools for the model to use	None
**kwargs	Any	Additional provider-specific parameters	{}

Returns:

Type	Description
CompletionResponse	CompletionResponse with generated content

Raises:

Type	Description
ProviderError	If the request fails

stream(messages: List[Message], *, model: str, temperature: float = 0.7, max_tokens: int = 4096, tools: Optional[List[ToolDefinition]] = None, **kwargs: Any) -> AsyncIterator[StreamChunk] abstractmethod async

Stream a chat completion response.

Parameters:

Name	Type	Description	Default
messages	List[Message]	List of conversation messages	required
model	str	Model identifier	required
temperature	float	Sampling temperature (0-2)	0.7
max_tokens	int	Maximum tokens to generate	4096
tools	Optional[List[ToolDefinition]]	Available tools for the model to use	None
**kwargs	Any	Additional provider-specific parameters	{}

Yields:

Type	Description
AsyncIterator[StreamChunk]	StreamChunk objects with incremental content

Raises:

Type	Description
ProviderError	If the request fails

discover_capabilities(model: str) -> ProviderRuntimeCapabilities async

Discover capabilities for the given model.

Default implementation falls back to configured limits and
provider-declared support flags. Providers should override
with real HTTP-based discovery when available.

stream_chat(messages: List[Message], *, model: str, temperature: float = 0.7, max_tokens: int = 4096, tools: Optional[List[ToolDefinition]] = None, **kwargs: Any) -> AsyncIterator[StreamChunk] async

Stream a chat completion response (alias for stream()).

This is an alias for the stream() method, provided for compatibility
with different naming conventions across SDKs (OpenAI uses stream,
Anthropic uses stream_chat, etc.).

Parameters:

Name	Type	Description	Default
messages	List[Message]	List of conversation messages	required
model	str	Model identifier	required
temperature	float	Sampling temperature (0-2)	0.7
max_tokens	int	Maximum tokens to generate	4096
tools	Optional[List[ToolDefinition]]	Available tools for the model to use	None
**kwargs	Any	Additional provider-specific parameters	{}

Yields:

Type	Description
AsyncIterator[StreamChunk]	StreamChunk objects with incremental content

count_tokens(text: str) -> int async

Estimate token count for given text.

Uses the fast native token counter when available and falls back
to word-based estimation.

Parameters:

Name	Type	Description	Default
text	str	Text to count tokens for	required

Returns:

Type	Description
int	Estimated token count

context_window(model: str) -> int

Get context window size for a given model.

Provides context window limits for common models to enable
context-budgeted tool selection strategies. Returns safe default
for unknown models.

Parameters:

Name	Type	Description	Default
model	str	Model identifier (e.g., "claude-sonnet-4-20250514", "qwen2.5-coder:7b")	required

Returns:

Type	Description
int	Context window in tokens. Returns safe default (8192) for unknown models.

Examples:

>>> provider = AnthropicProvider(api_key="...")
>>> cw = provider.context_window("claude-sonnet-4-20250514")
>>> assert cw == 200000

close() -> None abstractmethod async

Close any open connections or resources.

is_circuit_open() -> bool

Check if the circuit breaker is open (failing fast).

Returns:

Type	Description
bool	True if circuit is open and requests will be rejected

reset_circuit_breaker() -> None

Manually reset the circuit breaker to closed state.

BaseTool

victor.tools.base.BaseTool

Bases: ABC

Abstract base class for all tools.

Tools should implement name, description, parameters, and execute().
Optionally, tools can override the metadata property to provide
semantic information for dynamic tool selection.

name: str abstractmethod property

Tool name.

description: str abstractmethod property

Tool description.

parameters: Dict[str, Any] abstractmethod property

JSON Schema for tool parameters.

metadata: Optional[ToolMetadata] property

Semantic metadata for tool selection.

Override this property to provide category, keywords, use_cases,
and examples for dynamic tool selection. If None, metadata will
be auto-generated from tool properties.

Returns:

Type	Description
Optional[ToolMetadata]	ToolMetadata with semantic information, or None for auto-generation

cost_tier: CostTier property

Cost tier for the tool.

Override this property in subclasses to specify the appropriate tier.

Tiers

Returns:

Type	Description
CostTier	CostTier enum value

is_idempotent: bool property

Whether the tool execution is idempotent.

An idempotent tool produces the same result for the same input and has
no side effects that would change subsequent executions. This property
enables optimizations such as:
- Result caching (memoization)
- Safe retries on transient failures
- Parallel execution without coordination
- Deduplication of redundant calls

Examples of idempotent tools:
- File read operations
- Search/query operations
- Git status/log/diff (read-only)
- Web fetch (GET requests)

Examples of non-idempotent tools:
- File write/edit operations
- Git commit/push
- API mutations (POST, PUT, DELETE)
- Docker container operations

Override this property in subclasses for idempotent operations.

Returns:

Type	Description
bool	True if tool execution is idempotent, False otherwise (default)

get_metadata() -> ToolMetadata

Get semantic metadata for tool selection (ToolMetadataProvider contract).

This method fulfills the ToolMetadataProvider protocol and ALWAYS
returns valid ToolMetadata. It follows a two-tier strategy:

1. If explicit metadata is defined (via metadata property), use it
2. Otherwise, auto-generate metadata from tool properties

This ensures ALL tools can participate in semantic tool selection
without requiring manual configuration files.

Returns:

Type	Description
ToolMetadata	ToolMetadata with semantic information for tool selection

convert_parameters_to_schema(parameters: List[ToolParameter]) -> Dict[str, Any] staticmethod

Convert list of ToolParameter objects to JSON Schema format.

Parameters:

Name	Type	Description	Default
parameters	List[ToolParameter]	List of ToolParameter objects	required

Returns:

Type	Description
Dict[str, Any]	JSON Schema dictionary

execute(_exec_ctx: Dict[str, Any], **kwargs: Any) -> ToolResult abstractmethod async

Execute the tool.

Parameters:

Name	Type	Description	Default
_exec_ctx	Dict[str, Any]	Framework execution context (reserved name to avoid collision with tool parameters). Contains shared resources like code_manager.	required
**kwargs	Any	Tool parameters	{}

Returns:

Type	Description
ToolResult	ToolResult with execution outcome

to_json_schema() -> Dict[str, Any]

Convert tool to JSON Schema format.

Returns:

Type	Description
Dict[str, Any]	JSON Schema representation

to_schema(level: SchemaLevel = None) -> Dict[str, Any]

Generate JSON schema at specified verbosity level.

Parameters:

Name	Type	Description	Default
level	SchemaLevel	Schema verbosity level (FULL, COMPACT, or STUB). Defaults to FULL.	None

Returns:

Type	Description
Dict[str, Any]	JSON Schema with appropriate detail level.

Example

FULL: Complete schema (~100-150 tokens)

tool.to_schema(SchemaLevel.FULL)

COMPACT: All params, shorter descriptions (~60-80 tokens, ~20% reduction)

tool.to_schema(SchemaLevel.COMPACT)

STUB: Minimal schema, required params only (~25-40 tokens)

tool.to_schema(SchemaLevel.STUB)

validate_parameters(**kwargs: Any) -> bool

Validate provided parameters against schema.

Simple boolean validation - use validate_parameters_detailed() for
detailed error information.

Parameters:

Name	Type	Description	Default
**kwargs	Any	Parameters to validate	{}

Returns:

Type	Description
bool	True if valid, False otherwise

validate_parameters_detailed(**kwargs: Any) -> ToolValidationResult

Validate provided parameters against JSON Schema with detailed errors.

Uses JSON Schema Draft 7 validation for comprehensive type checking,
required field validation, enum constraints, and nested object validation.

Parameters:

Name	Type	Description	Default
**kwargs	Any	Parameters to validate	{}

Returns:

Type	Description
ToolValidationResult	ToolValidationResult with detailed error information