Tools API

from marsys.environment.utils import generate_openai_tool_schema

def generate_openai_tool_schema(
    func: Callable,
    func_name: str
) -> Dict[str, Any]

def search_web(query: str, max_results: int = 5) -> List[Dict]:
    """
    Search the web for information.

    Args:
        query: Search query string
        max_results: Maximum number of results to return

    Returns:
        List of search results
    """
    # Implementation
    pass

# Generate schema
schema = generate_openai_tool_schema(search_web, "search_web")

# Result:
{
    "type": "function",
    "function": {
        "name": "search_web",
        "description": "Search the web for information.",
        "parameters": {
            "type": "object",
            "properties": {
                "query": {
                    "type": "string",
                    "description": "Search query string"
                },
                "max_results": {
                    "type": "integer",
                    "description": "Maximum number of results to return",
                    "default": 5
                }
            },
            "required": ["query"]
        }
    }
}

def calculate_statistics(
    data: List[float],
    include_std: bool = True
) -> Dict[str, float]:
    """
    Calculate statistics for numerical data.

    Args:
        data: List of numerical values
        include_std: Whether to include standard deviation

    Returns:
        Dictionary with statistical measures
    """
    import statistics

    result = {
        "mean": statistics.mean(data),
        "median": statistics.median(data),
        "min": min(data),
        "max": max(data)
    }

    if include_std and len(data) > 1:
        result["std"] = statistics.stdev(data)

    return result

from typing import List, Dict, Optional, Literal

def process_data(
    input_data: Dict[str, Any],
    operation: Literal["transform", "filter", "aggregate"],
    options: Optional[Dict[str, Any]] = None
) -> Dict[str, Any]:
    """
    Process data with specified operation.

    Args:
        input_data: Input data dictionary
        operation: Type of operation to perform
        options: Additional operation options

    Returns:
        Processed data result
    """
    # Implementation
    pass

# Schema includes enum for operation
schema = generate_openai_tool_schema(process_data, "process_data")

async def fetch_api_data(
    endpoint: str,
    params: Optional[Dict[str, str]] = None,
    timeout: int = 30
) -> Dict[str, Any]:
    """
    Fetch data from API endpoint.

    Args:
        endpoint: API endpoint URL
        params: Query parameters
        timeout: Request timeout in seconds

    Returns:
        API response data
    """
    import aiohttp

    async with aiohttp.ClientSession() as session:
        async with session.get(
            endpoint,
            params=params,
            timeout=aiohttp.ClientTimeout(total=timeout)
        ) as response:
            return await response.json()

from marsys.agents import Agent

# Define tools
def search_tool(query: str) -> List[str]:
    """Search for information."""
    # Implementation
    return ["result1", "result2"]

def calculate_tool(expression: str) -> float:
    """Calculate mathematical expression."""
    # Implementation
    return eval(expression)  # Simplified example

# Create agent with tools
agent = Agent(
    model_config=model_config,
    name="Assistant",
    tools=[search_tool, calculate_tool],  # Auto-generates schemas
    goal="Assistant with search and calculation capabilities"
)

# Tools are automatically available to the agent

# Define tool with manual schema
manual_schema = {
    "type": "function",
    "function": {
        "name": "custom_tool",
        "description": "Custom tool with specific schema",
        "parameters": {
            "type": "object",
            "properties": {
                "input": {"type": "string"},
                "mode": {
                    "type": "string",
                    "enum": ["fast", "accurate"]
                }
            },
            "required": ["input"]
        }
    }
}

# Add to agent
agent = Agent(
    model_config=model_config,
    name="CustomAgent",
    tool_schemas=[manual_schema],  # Provide schemas directly
    tool_functions={"custom_tool": custom_tool_func}
)

from marsys.coordination.execution import ToolExecutor

async def execute_tool_call(
    tool_call: Dict[str, Any],
    available_tools: Dict[str, Callable],
    context: Dict[str, Any]
) -> Any

tool_call = {
    "id": "call_123",
    "type": "function",
    "function": {
        "name": "search_web",
        "arguments": '{"query": "AI trends"}'
    }
}

def read_csv_file(
    filepath: str,
    encoding: str = "utf-8",
    delimiter: str = ","
) -> List[Dict[str, str]]:
    """
    Read CSV file and return as list of dictionaries.

    Args:
        filepath: Path to CSV file
        encoding: File encoding
        delimiter: CSV delimiter

    Returns:
        List of row dictionaries
    """
    import csv

    with open(filepath, 'r', encoding=encoding) as f:
        reader = csv.DictReader(f, delimiter=delimiter)
        return list(reader)

import os
import requests

def get_weather(
    city: str,
    units: Literal["metric", "imperial"] = "metric"
) -> Dict[str, Any]:
    """
    Get current weather for a city.

    Args:
        city: City name
        units: Temperature units

    Returns:
        Weather data dictionary
    """
    api_key = os.getenv("WEATHER_API_KEY")
    url = "https://api.openweathermap.org/data/2.5/weather"

    response = requests.get(url, params={
        "q": city,
        "units": units,
        "appid": api_key
    })

    if response.status_code == 200:
        return response.json()
    else:
        raise Exception(f"Weather API error: {response.status_code}")

class StatefulTool:
    """Tool that maintains state between calls."""

    def __init__(self):
        self.history = []
        self.cache = {}

    def process_with_memory(
        self,
        input_data: str,
        use_cache: bool = True
    ) -> Dict[str, Any]:
        """
        Process data with memory of previous calls.

        Args:
            input_data: Input to process
            use_cache: Whether to use cached results

        Returns:
            Processing result with context
        """
        # Check cache
        if use_cache and input_data in self.cache:
            return {
                "result": self.cache[input_data],
                "cached": True,
                "history_length": len(self.history)
            }

        # Process
        result = self._process(input_data)

        # Update state
        self.history.append(input_data)
        self.cache[input_data] = result

        return {
            "result": result,
            "cached": False,
            "history_length": len(self.history)
        }

# Create instance and use as tool
stateful_tool = StatefulTool()
agent = Agent(
    model_config=config,
    tools=[stateful_tool.process_with_memory]
)