engine.function_call

class agentopera.engine.function_call.Tool(*args, **kwargs)

Bases: Protocol

property name: str

property description: str

property schema: ToolSchema

args_type() → Type[BaseModel]

return_type() → Type[Any]

state_type() → Type[BaseModel] | None

return_value_as_string(value: Any) → str

async run_json(args: Mapping[str, Any], cancellation_token: CancellationToken) → Any

save_state_json() → Mapping[str, Any]

load_state_json(state: Mapping[str, Any]) → None

class agentopera.engine.function_call.ToolSchema

Bases: TypedDict

parameters: NotRequired[ParametersSchema]

name: str

description: NotRequired[str]

strict: NotRequired[bool]

class agentopera.engine.function_call.ParametersSchema

Bases: TypedDict

type: str

properties: Dict[str, Any]

required: NotRequired[Sequence[str]]

additionalProperties: NotRequired[bool]

class agentopera.engine.function_call.BaseTool(args_type: Type[ArgsT], return_type: Type[ReturnT], name: str, description: str, strict: bool = False)

Bases: ABC, Tool, Generic[ArgsT, ReturnT]

property schema: ToolSchema

property name: str

property description: str

args_type() → Type[BaseModel]

return_type() → Type[Any]

state_type() → Type[BaseModel] | None

return_value_as_string(value: Any) → str

abstract async run(args: ArgsT, cancellation_token: CancellationToken) → ReturnT

async run_json(args: Mapping[str, Any], cancellation_token: CancellationToken) → Any

save_state_json() → Mapping[str, Any]

load_state_json(state: Mapping[str, Any]) → None

class agentopera.engine.function_call.BaseToolWithState(args_type: Type[ArgsT], return_type: Type[ReturnT], state_type: Type[StateT], name: str, description: str)

Bases: BaseTool[ArgsT, ReturnT], ABC, Generic[ArgsT, ReturnT, StateT]

abstract save_state() → StateT

abstract load_state(state: StateT) → None

save_state_json() → Mapping[str, Any]

load_state_json(state: Mapping[str, Any]) → None

class agentopera.engine.function_call.FunctionTool(func: Callable[[...], Any], description: str, name: str | None = None, global_imports: Sequence[str | ImportFromModule | Alias] = [], strict: bool = False)

Bases: BaseTool[BaseModel, BaseModel]

Create custom tools by wrapping standard Python functions.

FunctionTool offers an interface for executing Python functions either asynchronously or synchronously. Each function must include type annotations for all parameters and its return type. These annotations enable FunctionTool to generate a schema necessary for input validation, serialization, and for informing the LLM about expected parameters. When the LLM prepares a function call, it leverages this schema to generate arguments that align with the function’s specifications.

Note

It is the user’s responsibility to verify that the tool’s output type matches the expected type.

Parameters:

• func (Callable[..., ReturnT | Awaitable[ReturnT]]) – The function to wrap and expose as a tool.

• description (str) – A description to inform the model of the function’s purpose, specifying what it does and the context in which it should be called.

• name (str, optional) – An optional custom name for the tool. Defaults to the function’s original name if not provided.

• strict (bool, optional) – If set to True, the tool schema will only contain arguments that are explicitly defined in the function signature, and no default values will be allowed. Defaults to False. This is required to be set to True when used with models in structured output mode.

Example

import random
from agentopera.core import CancellationToken
from agentopera.core.tools import FunctionTool
from typing_extensions import Annotated
import asyncio


async def get_stock_price(ticker: str, date: Annotated[str, "Date in YYYY/MM/DD"]) -> float:
    # Simulates a stock price retrieval by returning a random float within a specified range.
    return random.uniform(10, 200)


async def example():
    # Initialize a FunctionTool instance for retrieving stock prices.
    stock_price_tool = FunctionTool(get_stock_price, description="Fetch the stock price for a given ticker.")

    # Execute the tool with cancellation support.
    cancellation_token = CancellationToken()
    result = await stock_price_tool.run_json({"ticker": "AAPL", "date": "2021/01/01"}, cancellation_token)

    # Output the result as a formatted string.
    print(stock_price_tool.return_value_as_string(result))


asyncio.run(example())

async run(args: BaseModel, cancellation_token: CancellationToken) → Any

class agentopera.engine.function_call.FunctionWithRequirements(func: 'Callable[P, T]', python_packages: 'Sequence[str]' = <factory>, global_imports: 'Sequence[Import]' = <factory>)

Bases: Generic[T, P]

func: Callable[[P], T]

python_packages: Sequence[str]

global_imports: Sequence[str | ImportFromModule | Alias]

classmethod from_callable(func: Callable[[P], T], python_packages: Sequence[str] = [], global_imports: Sequence[str | ImportFromModule | Alias] = []) → FunctionWithRequirements[T, P]

static from_str(func: str, python_packages: Sequence[str] = [], global_imports: Sequence[str | ImportFromModule | Alias] = []) → FunctionWithRequirementsStr

class agentopera.engine.function_call.FunctionWithRequirementsStr(func: 'str', python_packages: 'Sequence[str]' = [], global_imports: 'Sequence[Import]' = [])

Bases: object

func: str

python_packages: Sequence[str]

global_imports: Sequence[str | ImportFromModule | Alias]

compiled_func: Callable[[...], Any]

class agentopera.engine.function_call.FunctionCall(id: 'str', arguments: 'str', name: 'str')

Bases: object

id: str

arguments: str

name: str