calute.types.agent_types

calute.types.agent_types#

Agent type definitions for the Calute framework.

This module provides Pydantic-based data models for defining AI agents and their associated types within the Calute framework. It includes:

Agent: Core agent model with function calling, sampling parameters, and capabilities
AgentBaseFn: Abstract base class for defining class-based agent functions
AgentFunction: Type alias for callable agent functions
Response: Container for agent response data and conversation history
Result: Encapsulation of agent function return values

The agent types support features like: - Function calling with configurable strategies and timeouts - LLM sampling parameter configuration (temperature, top_p, etc.) - Automatic context compaction for long conversations - Agent switching triggers for multi-agent workflows - MCP (Model Context Protocol) server integration - Capability-based agent specialization

Typical usage example:

>>> from calute.types.agent_types import Agent
>>> agent = Agent(
...     name="Assistant",
...     model="gpt-4",
...     instructions="You are a helpful assistant.",
...     temperature=0.7,
...     functions=[my_tool_function],
... )
>>> agent.add_capability(AgentCapability(name="code_execution"))

class calute.types.agent_types.Agent(*, model: str | None = None, id: str | None = None, name: str | None = None, instructions: ~typing.Optional[~typing.Union[str, ~typing.Callable[[], str]]] = None, rules: ~typing.Optional[~typing.Union[list[str], ~typing.Callable[[], list[str]]]] = None, examples: list[str] | None = None, functions: list[typing.Union[typing.Callable, calute.types.agent_types.AgentBaseFn]] = <factory>, capabilities: list[calute.types.function_execution_types.AgentCapability] = <factory>, function_call_strategy: ~calute.types.function_execution_types.FunctionCallStrategy = FunctionCallStrategy.SEQUENTIAL, tool_choice: str | list[str] = None, parallel_tool_calls: bool = True, function_timeout: float | None = 30.0, max_function_retries: int = 3, top_p: float = 0.95, max_tokens: int = 2048, temperature: float = 0.7, top_k: int = 0, min_p: float = 0.0, presence_penalty: float = 0.0, frequency_penalty: float = 0.0, repetition_penalty: float = 1.0, extra_body: dict | None = None, stop: str | list[str] | None = None, auto_compact: bool = False, compact_threshold: float = 0.8, compact_target: float = 0.5, max_context_tokens: int | None = None, compaction_strategy: ~calute.types.function_execution_types.CompactionStrategy = CompactionStrategy.SUMMARIZE, preserve_system_prompt: bool = True, preserve_recent_messages: int = 5, switch_triggers: list[calute.types.function_execution_types.AgentSwitchTrigger] = <factory>, fallback_agent_id: str | None = None)[source]#

Bases: BaseModel

Core agent model with function calling and switching capabilities.

Agent is the primary Pydantic model for defining AI agents in the Calute framework. It encapsulates all configuration needed for an agent including its identity, instructions, available functions/tools, LLM sampling parameters, and advanced features like context compaction and agent switching.

The Agent class supports both simple single-agent use cases and complex multi-agent workflows with dynamic agent switching based on triggers.

model#

LLM model identifier (e.g., ‘gpt-4’, ‘claude-3’). If None, uses the default model from the Calute instance.

Type: str | None

id#

Unique identifier for the agent, used in multi-agent routing.

Type: str | None

name#

Human-readable name for the agent.

Type: str | None

instructions#

System prompt or callable returning system prompt text.

Type: Optional[Union[str, Callable[[], str]]]

rules#

List of rules or callable returning rules the agent must follow.

Type: Optional[Union[list[str], Callable[[], list[str]]]]

examples#

List of example interactions for few-shot learning.

Type: list[str] | None

functions#

List of callable functions/tools available to the agent.

Type: list[Union[Callable, calute.types.agent_types.AgentBaseFn]]

capabilities#

List of AgentCapability objects defining special abilities.

Type: list[calute.types.function_execution_types.AgentCapability]

function_call_strategy#

Strategy for executing multiple function calls (SEQUENTIAL, PARALLEL, etc.).

Type: calute.types.function_execution_types.FunctionCallStrategy

tool_choice#

Specific tool(s) the agent should prefer using.

Type: str | list[str]

parallel_tool_calls#

Whether to allow parallel tool execution.

Type: bool

function_timeout#

Timeout in seconds for individual function calls.

Type: float | None

max_function_retries#

Maximum retry attempts for failed function calls.

Type: int

top_p#

Nucleus sampling parameter (0.0-1.0).

Type: float

max_tokens#

Maximum tokens in generated responses.

Type: int

temperature#

Sampling temperature controlling randomness (0.0-2.0).

Type: float

top_k#

Top-k sampling parameter (0 disables).

Type: int

min_p#

Minimum probability threshold for sampling.

Type: float

presence_penalty#

Penalty for token presence (-2.0 to 2.0).

Type: float

frequency_penalty#

Penalty for token frequency (-2.0 to 2.0).

Type: float

repetition_penalty#

Multiplicative penalty for repetition.

Type: float

extra_body#

Additional parameters passed to the LLM API.

Type: dict | None

stop#

Stop sequences that halt generation.

Type: str | list[str] | None

auto_compact#

Whether to automatically compact context when near limit.

Type: bool

compact_threshold#

Context usage ratio triggering compaction (0.0-1.0).

Type: float

compact_target#

Target context usage ratio after compaction (0.0-1.0).

Type: float

max_context_tokens#

Maximum context tokens before compaction.

Type: int | None

compaction_strategy#

Strategy for compacting conversation history.

Type: calute.types.function_execution_types.CompactionStrategy

preserve_system_prompt#

Whether to preserve system prompt during compaction.

Type: bool

preserve_recent_messages#

Number of recent messages to preserve.

Type: int

switch_triggers#

List of triggers for switching to other agents.

Type: list[calute.types.function_execution_types.AgentSwitchTrigger]

fallback_agent_id#

Agent ID to switch to when no triggers match.

Type: str | None

Example

>>> agent = Agent(
...     name="CodeAssistant",
...     model="gpt-4",
...     instructions="You are a helpful coding assistant.",
...     temperature=0.3,
...     functions=[search_docs, run_code],
...     max_tokens=4096,
... )
>>> agent.set_sampling_params(top_p=0.9)
>>> print(agent.get_available_functions())
['search_docs', 'run_code']

add_capability(capability: AgentCapability) → None[source]#

Add a capability to the agent’s capability list.

Capabilities define special abilities or permissions for the agent, such as code execution, file access, or web browsing.

Parameters: capability – An AgentCapability object defining the capability to add to this agent.

Example

>>> agent.add_capability(AgentCapability(
...     name="code_execution",
...     description="Can execute Python code"
... ))

attach_mcp(mcp_servers: MCPManager | MCPServerConfig | list, server_names: list[str] | None = None) → None[source]#

Attach MCP servers to this agent, connecting and adding their tools.

This method provides a convenient way to connect MCP servers and automatically add their tools to the agent’s function list.

Parameters

mcp_servers – Can be one of: - MCPManager: An existing MCP manager instance - MCPServerConfig: A single server config (will create manager and connect) - list[MCPServerConfig]: Multiple server configs (will create manager and connect all)
server_names – Optional list of server names to filter tools from. If None, adds tools from all servers in the manager.

Example

>>>
>>> agent.attach_mcp(MCPServerConfig(
...     name="filesystem",
...     command="npx",
...     args=["-y", "@modelcontextprotocol/server-filesystem", "/tmp"]
... ))
>>>
>>>
>>> agent.attach_mcp([
...     MCPServerConfig(name="filesystem", ...),
...     MCPServerConfig(name="sqlite", ...)
... ])
>>>
>>>
>>> manager = MCPManager()
>>> await manager.add_server(config1)
>>> await manager.add_server(config2)
>>> agent.attach_mcp(manager, server_names=["filesystem"])

auto_compact: bool#

capabilities: list[AgentCapability]#

compact_target: float#

compact_threshold: float#

compaction_strategy: CompactionStrategy#

examples: list[str] | None#

extra_body: dict | None#

fallback_agent_id: str | None#

frequency_penalty: float#

function_call_strategy: FunctionCallStrategy#

function_timeout: float | None#

functions: list[tp.Callable | AgentBaseFn]#

get_available_functions() → list[str][source]#

Get a list of all available function/tool names.

Returns the names of all functions registered with this agent, useful for introspection and debugging.

Returns: A list of function name strings.

Example

>>> print(agent.get_available_functions())
['search_docs', 'execute_code', 'send_email']

get_functions_mapping() → dict[str, Callable][source]#

Get a mapping of function names to their callable objects.

Creates a dictionary for quick lookup of function objects by name, useful for dynamic function invocation and introspection.

Returns: A dictionary mapping function names (str) to their callable implementations.

Example

>>> mapping = agent.get_functions_mapping()
>>> search_fn = mapping.get('search_docs')
>>> result = search_fn(query="python tutorials")

has_capability(capability_name: str) → bool[source]#

Check if the agent has a specific capability by name.

Useful for conditional logic based on agent capabilities in multi-agent systems or capability-gated operations.

Parameters: capability_name – The name of the capability to check for.
Returns: True if the agent has a capability with the specified name, False otherwise.

Example

>>> if agent.has_capability("code_execution"):
...     result = execute_code(code_snippet)

id: str | None#

instructions: str | tp.Callable[[], str] | None#

max_context_tokens: int | None#

max_function_retries: int#

max_tokens: int#

min_p: float#

model: str | None#

model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

name: str | None#

parallel_tool_calls: bool#

presence_penalty: float#

preserve_recent_messages: int#

preserve_system_prompt: bool#

repetition_penalty: float#

rules: list[str] | tp.Callable[[], list[str]] | None#

set_model(model_id: str) → None[source]#

Set the LLM model identifier for this agent.

Updates the agent’s model configuration to use the specified model. This is useful for dynamically switching models at runtime.

Parameters: model_id – The model identifier string (e.g., ‘gpt-4’, ‘claude-3-opus’).

Update LLM sampling parameters for text generation.

Allows selective updating of sampling parameters. Only parameters that are explicitly provided (not None) will be updated; others retain their current values.

Parameters

top_p – Nucleus sampling probability threshold (0.0-1.0). Higher values include more tokens in sampling.
max_tokens – Maximum number of tokens to generate.
temperature – Controls randomness in generation (0.0-2.0). Lower values make output more deterministic.
top_k – Number of highest probability tokens to consider. 0 disables top-k sampling.
min_p – Minimum probability threshold for token consideration.
presence_penalty – Penalty for tokens already present (-2.0 to 2.0). Positive values encourage new topics.
frequency_penalty – Penalty based on token frequency (-2.0 to 2.0). Positive values reduce repetition.
repetition_penalty – Multiplicative penalty for repeated tokens. Values > 1.0 reduce repetition.

Example

>>> agent.set_sampling_params(
...     temperature=0.5,
...     top_p=0.9,
...     max_tokens=2048
... )

stop: str | list[str] | None#

switch_triggers: list[AgentSwitchTrigger]#

temperature: float#

tool_choice: str | list[str]#

top_k: int#

top_p: float#

class calute.types.agent_types.Result(*, value: str = '', agent: calute.types.agent_types.Agent | None = None, context_variables: dict = <factory>)[source]#

Bases: BaseModel

Encapsulates return values from agent function/tool calls.

Result provides a structured way for agent functions to return not just a value, but also control flow information (like switching agents) and context updates. This enables sophisticated multi-agent workflows where tools can influence agent behavior.

Functions can return a Result to: - Provide a string value to be included in the conversation - Trigger an agent switch by specifying a new agent - Update context variables for subsequent processing

value#

The string result to be included in the conversation as the function’s output. Defaults to empty string.

Type: str

agent#

Optional Agent instance to switch to after this function completes. Used for dynamic agent handoffs.

Type: calute.types.agent_types.Agent | None

context_variables#

Dictionary of variables to merge into the conversation context, persisting across turns.

Type: dict

Example

>>> def transfer_to_specialist(context: dict) -> Result:
...     specialist = Agent(name="Specialist", ...)
...     return Result(
...         value="Transferring you to our specialist.",
...         agent=specialist,
...         context_variables={"transferred": True}
...     )

agent: Agent | None#

context_variables: dict#

model_config: ClassVar[ConfigDict] = {}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

value: str#

calute.types.agent_types

Contents

calute.types.agent_types#