calute.types.function_execution_types

Type definitions for function execution and streaming responses.

This module provides comprehensive type definitions for the function execution system in Calute, including: - Execution strategies (sequential, parallel, pipeline) - Agent switching triggers and capabilities - Function call request and result types - Streaming response chunk types - Execution status and completion types

These types form the foundation for type-safe function execution, agent orchestration, and streaming response handling throughout the Calute framework.

Example

>>> from calute.types.function_execution_types import (
...     ExecutionStatus,
...     RequestFunctionCall,
...     StreamChunk,
... )
>>> call = RequestFunctionCall(name="search", arguments={"query": "test"})
>>> call.status = ExecutionStatus.SUCCESS

class calute.types.function_execution_types.AgentCapability(name: str, description: str, function_names: list[str] = <factory>, context_requirements: dict[str, typing.Any] = <factory>, performance_score: float = 1.0)

Bases: object

Definition of an agent’s capability.

Describes a specific capability that an agent possesses, including the functions it can execute and performance metrics.

name

Human-readable name of the capability.

Type

str

description

Detailed description of what the capability enables.

Type

str

function_names

List of function names associated with this capability.

Type

list[str]

context_requirements

Required context variables for this capability.

Type

dict[str, Any]

performance_score

Performance rating for this capability (0.0-1.0).

Type

float

FUNCTION_CALLING = AgentCapability(name='function_calling', description='Can use tools and function calls', function_names=[], context_requirements={}, performance_score=1.0)

context_requirements: dict[str, Any]

description: str

function_names: list[str]

name: str

performance_score: float = 1.0

class calute.types.function_execution_types.AgentSwitch(type: str = 'agent_switch', from_agent: str = '', to_agent: str = '', reason: str = '')

Bases: object

Event notification for an agent switch.

Emitted when execution switches from one agent to another, recording the transition details.

type

Event type identifier (default: “agent_switch”).

Type

str

from_agent

ID of the agent being switched from.

Type

str

to_agent

ID of the agent being switched to.

Type

str

reason

Human-readable reason for the switch.

Type

str

from_agent: str = ''

reason: str = ''

to_agent: str = ''

type: str = 'agent_switch'

class calute.types.function_execution_types.AgentSwitchTrigger(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)

Bases: Enum

Enumeration of triggers for agent switching.

Defines the conditions or events that can trigger switching from one agent to another during execution.

EXPLICIT

Manual/explicit switch requested by code or user.

CAPABILITY_BASED

Switch based on required capabilities.

LOAD_BALANCING

Switch to balance load across agents.

CONTEXT_BASED

Switch based on conversation context.

ERROR_RECOVERY

Switch triggered by error recovery mechanism.

CAPABILITY_BASED = 'capability'

CAPABILITY_REQUIRED = 'capability'

CONTEXT_BASED = 'context'

CUSTOM = 'custom'

ERROR_RECOVERY = 'error'

EXPLICIT = 'explicit'

LOAD_BALANCING = 'load'

class calute.types.function_execution_types.CompactionStrategy(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)

Bases: Enum

Enumeration of strategies for context compaction.

Defines how conversation context should be reduced when it exceeds token limits or needs optimization.

SUMMARIZE

Compress context by generating a summary.

SLIDING_WINDOW

Keep only recent context within a window.

PRIORITY_BASED

Keep high-priority context, remove low-priority.

TRUNCATE

Simply truncate context to fit limits.

PRIORITY_BASED = 'priority_based'

SLIDING_WINDOW = 'sliding_window'

SUMMARIZE = 'summarize'

TRUNCATE = 'truncate'

class calute.types.function_execution_types.Completion(type: str = 'completion', final_content: str = '', reasoning_content: str = '', function_calls_executed: int = 0, agent_id: str = '', execution_history: list[typing.Any] = <factory>)

Bases: object

Final completion event for a response.

Emitted when the entire response cycle completes, including all function executions and agent responses.

type

Event type identifier (default: “completion”).

Type

str

final_content

The final accumulated response content.

Type

str

reasoning_content

Accumulated reasoning/thinking tokens from the model.

Type

str

function_calls_executed

Total number of function calls executed.

Type

int

agent_id

ID of the agent that produced the final response.

Type

str

execution_history

List of execution events that occurred.

Type

list[Any]

agent_id: str = ''

execution_history: list[Any]

final_content: str = ''

function_calls_executed: int = 0

reasoning_content: str = ''

type: str = 'completion'

class calute.types.function_execution_types.ExecutionResult(status: ExecutionStatus, result: Any | None = None, error: str | None = None)

Bases: object

Result container for function execution.

Holds the outcome of a function execution attempt, including status, result value, and any error information.

status

The execution status (success, failure, etc.).

Type

calute.types.function_execution_types.ExecutionStatus

result

The return value from successful execution.

Type

Any | None

error

Error message if execution failed.

Type

str | None

error: str | None = None

result: Any | None = None

status: ExecutionStatus

class calute.types.function_execution_types.ExecutionStatus(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)

Bases: Enum

Enumeration of function/agent execution status values.

Represents the current state or outcome of a function or agent execution operation.

SUCCESS

Execution completed successfully.

FAILURE

Execution failed with an error.

PARTIAL

Execution partially completed (some steps succeeded).

PENDING

Execution is pending/not yet started.

CANCELLED

Execution was cancelled before completion.

CANCELLED = 'cancelled'

FAILED = 'failure'

FAILURE = 'failure'

PARTIAL = 'partial'

PENDING = 'pending'

SUCCESS = 'success'

class calute.types.function_execution_types.FunctionCallInfo(name: str, id: str)

Bases: object

Basic identifying information for a function call.

Lightweight container for function call identification, used in event notifications and tracking.

name

Name of the function being called.

Type

str

id

Unique identifier for this specific call.

Type

str

id: str

name: str

class calute.types.function_execution_types.FunctionCallStrategy(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)

Bases: Enum

Enumeration of strategies for handling function calls.

Defines how multiple function calls should be executed when an agent needs to invoke several functions.

SEQUENTIAL

Execute functions one after another in order.

PARALLEL

Execute all functions concurrently.

CONDITIONAL

Execute functions based on conditions/dependencies.

PIPELINE

Execute functions in a pipeline where output flows to input.

CONDITIONAL = 'conditional'

PARALLEL = 'parallel'

PIPELINE = 'pipeline'

SEQUENTIAL = 'sequential'

class calute.types.function_execution_types.FunctionCallsExtracted(type: str = 'function_calls_extracted', function_calls: list[calute.types.function_execution_types.FunctionCallInfo] = <factory>, agent_id: str = '')

Bases: object

Event containing extracted function call information.

Emitted after function calls have been parsed and extracted from an LLM response, before execution begins.

type

Event type identifier (default: “function_calls_extracted”).

Type

str

function_calls

List of extracted function call information.

Type

list[calute.types.function_execution_types.FunctionCallInfo]

agent_id

ID of the agent that requested the function calls.

Type

str

agent_id: str = ''

function_calls: list[calute.types.function_execution_types.FunctionCallInfo]

type: str = 'function_calls_extracted'

class calute.types.function_execution_types.FunctionDetection(type: str = 'function_detection', message: str = '', agent_id: str = '')

Bases: object

Notification event for function call detection.

Emitted when the system detects that an LLM response contains function/tool calls that need to be executed.

type

Event type identifier (default: “function_detection”).

Type

str

message

Human-readable message about the detection.

Type

str

agent_id

ID of the agent whose response contained the calls.

Type

str

agent_id: str = ''

message: str = ''

type: str = 'function_detection'

class calute.types.function_execution_types.FunctionExecutionComplete(type: str = 'function_execution_complete', function_name: str = '', function_id: str = '', status: str = '', result: Any | None = None, error: str | None = None, agent_id: str = '')

Bases: object

Event notification for function execution completion.

Emitted when a function finishes execution, containing the result or error information.

type

Event type identifier (default: “function_execution_complete”).

Type

str

function_name

Name of the executed function.

Type

str

function_id

Unique identifier for this function call.

Type

str

status

Execution status string (e.g., “success”, “error”).

Type

str

result

Return value from successful execution.

Type

Any | None

error

Error message if execution failed.

Type

str | None

agent_id

ID of the agent that executed the function.

Type

str

agent_id: str = ''

error: str | None = None

function_id: str = ''

function_name: str = ''

result: Any | None = None

status: str = ''

type: str = 'function_execution_complete'

class calute.types.function_execution_types.FunctionExecutionStart(type: str = 'function_execution_start', function_name: str = '', function_id: str = '', progress: str = '', agent_id: str = '')

Bases: object

Event notification for function execution start.

Emitted when a function begins execution, allowing for progress tracking and monitoring.

type

Event type identifier (default: “function_execution_start”).

Type

str

function_name

Name of the function being executed.

Type

str

function_id

Unique identifier for this function call.

Type

str

progress

Progress indicator (e.g., “1/3” for first of three).

Type

str

agent_id

ID of the agent executing the function.

Type

str

agent_id: str = ''

function_id: str = ''

function_name: str = ''

progress: str = ''

type: str = 'function_execution_start'

class calute.types.function_execution_types.ReinvokeSignal(message: str, agent_id: str, type: str = 'reinvoke_signal')

Bases: object

Signal that the agent is being reinvoked with function results.

Emitted when the agent is called again after function execution, allowing it to process the function results and continue.

message

Informational message about the reinvocation.

Type

str

agent_id

ID of the agent being reinvoked.

Type

str

type

Event type identifier (default: “reinvoke_signal”).

Type

str

agent_id: str

message: str

type: str = 'reinvoke_signal'

class calute.types.function_execution_types.RequestFunctionCall(name: str, arguments: dict, id: str = <factory>, call_id: str | None = None, agent_id: str | None = None, dependencies: list[str] = <factory>, timeout: float | None = None, retry_count: int = 0, max_retries: int = 3, status: ~calute.types.function_execution_types.ExecutionStatus = ExecutionStatus.PENDING, result: ~typing.Any = None, error: str | None = None)

Bases: object

Enhanced representation of a function call request.

Encapsulates all information needed to execute a function call, including execution parameters, retry configuration, and status tracking.

name

Name of the function to call.

Type

str

arguments

Dictionary of arguments to pass to the function.

Type

dict

id

Unique identifier for this function call, auto-generated if not provided.

Type

str

call_id

Optional explicit call ID that, when provided, overrides id. After initialization, id and call_id are kept in sync.

Type

str | None

agent_id

ID of the agent making the call, if applicable.

Type

str | None

dependencies

List of function call IDs this call depends on.

Type

list[str]

timeout

Optional timeout in seconds for execution.

Type

float | None

retry_count

Current number of retry attempts made.

Type

int

max_retries

Maximum number of retry attempts allowed.

Type

int

status

Current execution status of the function call.

Type

calute.types.function_execution_types.ExecutionStatus

result

Result value from successful execution.

Type

Any

error

Error message if execution failed.

Type

str | None

agent_id: str | None = None

arguments: dict

call_id: str | None = None

dependencies: list[str]

error: str | None = None

id: str

max_retries: int = 3

name: str

result: Any = None

retry_count: int = 0

status: ExecutionStatus = 'pending'

timeout: float | None = None

class calute.types.function_execution_types.ResponseResult(content: str, response: ~openai.types.chat.chat_completion.ChatCompletion, completion: ~calute.types.function_execution_types.Completion, reasoning_content: str = '', function_calls: list[calute.types.function_execution_types.RequestFunctionCall] = <factory>, agent_id: str = '', execution_history: list[typing.Any] = <factory>, reinvoked: bool = False)

Bases: object

Complete result from a non-streaming response.

Contains all information from a completed non-streaming agent response, including content and execution details.

content

The text content of the response.

Type

str

response

Raw ChatCompletion response from the LLM.

Type

openai.types.chat.chat_completion.ChatCompletion

completion

Completion event with summary information.

Type

calute.types.function_execution_types.Completion

reasoning_content

Reasoning/thinking content produced by the model, if any.

Type

str

function_calls

List of function calls that were executed.

Type

list[calute.types.function_execution_types.RequestFunctionCall]

agent_id

ID of the agent that produced the response.

Type

str

execution_history

List of all execution events.

Type

list[Any]

reinvoked

Whether the agent was reinvoked after function execution.

Type

bool

agent_id: str = ''

completion: Completion

content: str

execution_history: list[Any]

function_calls: list[calute.types.function_execution_types.RequestFunctionCall]

reasoning_content: str = ''

reinvoked: bool = False

response: ChatCompletion

class calute.types.function_execution_types.StreamChunk(type: str = 'stream_chunk', chunk: openai.types.chat.chat_completion_chunk.ChatCompletionChunk | google.generativeai.types.generation_types.GenerateContentResponse | None = None, agent_id: str = '', content: str | None = None, buffered_content: str | None = None, reasoning_content: str | None = None, buffered_reasoning_content: str | None = None, function_calls_detected: bool | None = None, reinvoked: bool = False, tool_calls: list[calute.types.function_execution_types.ToolCallStreamChunk] | None = None, streaming_tool_calls: list[calute.types.function_execution_types.ToolCallStreamChunk] | None = None)

Bases: object

Streaming chunk response from an LLM provider.

Encapsulates a streaming response chunk, supporting both OpenAI and Gemini response formats with tool call tracking.

type

Type identifier for this chunk (default: “stream_chunk”).

Type

str

chunk

Raw chunk from OpenAI or Gemini API.

Type

openai.types.chat.chat_completion_chunk.ChatCompletionChunk | google.generativeai.types.generation_types.GenerateContentResponse | None

agent_id

ID of the agent that produced this chunk.

Type

str

content

Text content extracted from the chunk.

Type

str | None

buffered_content

Accumulated content from all chunks so far.

Type

str | None

reasoning_content

Reasoning/thinking content from the current chunk.

Type

str | None

buffered_reasoning_content

Accumulated reasoning content from all chunks so far.

Type

str | None

function_calls_detected

Whether function calls were detected.

Type

bool | None

reinvoked

Whether the agent was reinvoked after function execution.

Type

bool

tool_calls

Completed tool calls extracted from the response.

Type

list[calute.types.function_execution_types.ToolCallStreamChunk] | None

streaming_tool_calls

Tool calls still being streamed.

Type

list[calute.types.function_execution_types.ToolCallStreamChunk] | None

agent_id: str = ''

buffered_content: str | None = None

buffered_reasoning_content: str | None = None

chunk: openai.types.chat.chat_completion_chunk.ChatCompletionChunk | google.generativeai.types.generation_types.GenerateContentResponse | None = None

content: str | None = None

function_calls_detected: bool | None = None

property gemini_content: str | None

Extract content from a Gemini response chunk.

Returns

The text content from the Gemini response, or None if unavailable.

property is_thinking: bool

Check if currently inside thinking/reasoning tags.

Detects whether the buffered content is within thinking or reasoning XML-style tags (e.g., <think>, <reasoning>).

Returns

True if inside thinking tags, False otherwise.

reasoning_content: str | None = None

reinvoked: bool = False

streaming_tool_calls: list[calute.types.function_execution_types.ToolCallStreamChunk] | None = None

tool_calls: list[calute.types.function_execution_types.ToolCallStreamChunk] | None = None

type: str = 'stream_chunk'

class calute.types.function_execution_types.SwitchContext(function_results: list[calute.types.function_execution_types.ExecutionResult], execution_error: bool, buffered_content: str | None = None)

Bases: object

Context information for agent switching decisions.

Provides the context needed to make decisions about switching between agents during execution.

function_results

Results from recent function executions.

Type

list[calute.types.function_execution_types.ExecutionResult]

execution_error

Whether an error occurred during execution.

Type

bool

buffered_content

Any buffered response content.

Type

str | None

buffered_content: str | None = None

execution_error: bool

function_results: list[calute.types.function_execution_types.ExecutionResult]

class calute.types.function_execution_types.ToolCallStreamChunk(id: str, type: str = 'function', function_name: str | None = None, arguments: str | None = None, index: int | None = None, is_complete: bool = False)

Bases: object

Streaming chunk for a tool/function call.

Represents a partial chunk of a tool or function call received during streaming responses.

id

Unique identifier for this tool call.

Type

str

type

Type of tool call (default: “function”).

Type

str

function_name

Name of the function being called.

Type

str | None

arguments

JSON string of arguments (may be partial during streaming).

Type

str | None

index

Index of this tool call in the response.

Type

int | None

is_complete

Whether the tool call is complete.

Type

bool

arguments: str | None = None

function_name: str | None = None

id: str

index: int | None = None

is_complete: bool = False

type: str = 'function'