Tools API¶

Complete API reference for the Tool system.

Tool Base Class¶

Tool¶

class Tool(ABC):
    """Abstract base class for agent tools."""

Tools provide specific capabilities to agents. When an agent needs to perform an action beyond text generation, it uses tools.

Abstract Methods¶

All tools must implement these methods:

get_name¶

@abstractmethod
def get_name(self) -> str

Return the unique name of the tool.

Returns: str - Tool name (used by LLM to identify the tool)

get_description¶

@abstractmethod
def get_description(self) -> str

Return a description of what the tool does.

Returns: str - Description (used by LLM to decide when to use the tool)

get_input_schema¶

@abstractmethod
def get_input_schema(self) -> dict

Return the JSON Schema for tool inputs.

Returns: dict - JSON Schema object

execute¶

@abstractmethod
def execute(self, **kwargs) -> dict

Execute the tool with the provided arguments.

Parameters: - **kwargs - Tool arguments matching the input schema

Returns: dict - Tool execution result

Input Schema Format¶

The input schema follows JSON Schema specification:

Basic Schema¶

def get_input_schema(self) -> dict:
    return {
        "type": "object",
        "properties": {
            "param_name": {
                "type": "string",
                "description": "Parameter description"
            }
        },
        "required": ["param_name"]
    }

Supported Types¶

Type	JSON Schema	Python Type
String	`"type": "string"`	`str`
Integer	`"type": "integer"`	`int`
Number	`"type": "number"`	`float`
Boolean	`"type": "boolean"`	`bool`
Array	`"type": "array"`	`list`
Object	`"type": "object"`	`dict`

String with Constraints¶

{
    "type": "string",
    "minLength": 1,
    "maxLength": 100,
    "pattern": "^[a-zA-Z]+$",
    "description": "Alphabetic string"
}

Enum Values¶

{
    "type": "string",
    "enum": ["option1", "option2", "option3"],
    "description": "Select one option"
}

Number with Range¶

{
    "type": "integer",
    "minimum": 1,
    "maximum": 100,
    "description": "Number between 1 and 100"
}

Array Schema¶

{
    "type": "array",
    "items": {"type": "string"},
    "minItems": 1,
    "maxItems": 10,
    "description": "List of strings"
}

Nested Object¶

{
    "type": "object",
    "properties": {
        "name": {"type": "string"},
        "address": {
            "type": "object",
            "properties": {
                "street": {"type": "string"},
                "city": {"type": "string"}
            }
        }
    }
}

Built-in Tools¶

AkordiChatTool¶

class AkordiChatTool(Tool):
    """Tool for managing chat sessions and messages in DynamoDB."""

Operations¶

Operation	Description
`create_session`	Create a new chat session
`add_message`	Add a message to a session
`get_history`	Get chat history
`list_sessions`	List user's sessions
`get_session`	Get session details

Usage¶

from akordi_agents.tools import AkordiChatTool

chat_tool = AkordiChatTool()

# Create session
result = chat_tool.execute(
    operation="create_session",
    user_id="user-123",
    title="My Chat",
)

# Add message
result = chat_tool.execute(
    operation="add_message",
    chat_id="chat-456",
    actor="user",
    message="Hello!",
)

# Get history
result = chat_tool.execute(
    operation="get_history",
    chat_id="chat-456",
    limit=50,
)

Input Schema¶

{
    "type": "object",
    "properties": {
        "operation": {
            "type": "string",
            "enum": ["create_session", "add_message", "get_history",
                     "list_sessions", "get_session"],
            "description": "Operation to perform"
        },
        "user_id": {
            "type": "string",
            "description": "User identifier"
        },
        "chat_id": {
            "type": "string",
            "description": "Chat session ID"
        },
        "message": {
            "type": "string",
            "description": "Message content"
        },
        "actor": {
            "type": "string",
            "enum": ["user", "assistant", "system"],
            "description": "Message sender"
        },
        "title": {
            "type": "string",
            "description": "Session title"
        },
        "limit": {
            "type": "integer",
            "description": "Max results to return"
        }
    },
    "required": ["operation"]
}

Creating Custom Tools¶

Basic Example¶

from akordi_agents.tools import Tool

class WeatherTool(Tool):
    """Get current weather for a location."""

    def get_name(self) -> str:
        return "get_weather"

    def get_description(self) -> str:
        return "Get the current weather conditions for a city"

    def get_input_schema(self) -> dict:
        return {
            "type": "object",
            "properties": {
                "city": {
                    "type": "string",
                    "description": "City name"
                },
                "units": {
                    "type": "string",
                    "enum": ["celsius", "fahrenheit"],
                    "default": "celsius"
                }
            },
            "required": ["city"]
        }

    def execute(self, **kwargs) -> dict:
        city = kwargs.get("city")
        units = kwargs.get("units", "celsius")

        # Your implementation
        weather = self.fetch_weather(city, units)

        return {
            "success": True,
            "data": weather
        }

With External API¶

import requests

class APITool(Tool):
    def __init__(self, api_key: str, base_url: str):
        self.api_key = api_key
        self.base_url = base_url

    def get_name(self) -> str:
        return "api_tool"

    def get_description(self) -> str:
        return "Make API calls to external service"

    def get_input_schema(self) -> dict:
        return {
            "type": "object",
            "properties": {
                "endpoint": {"type": "string"},
                "method": {
                    "type": "string",
                    "enum": ["GET", "POST"]
                },
                "data": {"type": "object"}
            },
            "required": ["endpoint"]
        }

    def execute(self, **kwargs) -> dict:
        endpoint = kwargs.get("endpoint")
        method = kwargs.get("method", "GET")
        data = kwargs.get("data", {})

        headers = {"Authorization": f"Bearer {self.api_key}"}
        url = f"{self.base_url}/{endpoint}"

        try:
            if method == "GET":
                response = requests.get(url, headers=headers)
            else:
                response = requests.post(url, json=data, headers=headers)

            return {
                "success": True,
                "status_code": response.status_code,
                "data": response.json()
            }
        except Exception as e:
            return {
                "success": False,
                "error": str(e)
            }

With Validation¶

class ValidatedTool(Tool):
    def execute(self, **kwargs) -> dict:
        # Validate required fields
        required = ["field1", "field2"]
        missing = [f for f in required if not kwargs.get(f)]

        if missing:
            return {
                "success": False,
                "error": f"Missing required fields: {missing}"
            }

        # Validate field types
        if not isinstance(kwargs.get("field1"), str):
            return {
                "success": False,
                "error": "field1 must be a string"
            }

        # Execute logic
        return self.do_work(kwargs)

Async Tool¶

import asyncio

class AsyncTool(Tool):
    async def execute_async(self, **kwargs) -> dict:
        # Async implementation
        result = await self.async_operation(kwargs)
        return {"success": True, "data": result}

    def execute(self, **kwargs) -> dict:
        # Sync wrapper
        loop = asyncio.get_event_loop()
        return loop.run_until_complete(self.execute_async(**kwargs))

Tool Response Format¶

Success Response¶

{
    "success": True,
    "data": {
        # Tool-specific result data
    }
}

Error Response¶

{
    "success": False,
    "error": "Error message",
    "error_code": "ERROR_CODE",  # Optional
    "details": {}                 # Optional
}

Using Tools with Agents¶

With create_langgraph_agent¶

from akordi_agents.core import create_langgraph_agent

agent = create_langgraph_agent(
    name="tool_agent",
    llm_service=llm_service,
    tools=[
        WeatherTool(),
        CalculatorTool(),
        SearchTool(),
    ],
    config={"enable_tools": True}
)

response = agent.process_request({
    "query": "What's the weather in London?",
    "system_message": "Use available tools to answer questions.",
})

With AgentBuilder¶

from akordi_agents.core import AgentBuilder

agent = (
    AgentBuilder("my_agent")
    .with_llm_service_instance(llm_service)
    .with_tools([weather_tool, calculator_tool])
    .with_langgraph(enable=True, config={"enable_tools": True})
    .build()
)

Tool Execution in Workflows¶

Tools are executed by the ToolExecutionNode in LangGraph workflows:

sequenceDiagram
    participant LLM
    participant TD as ToolDecisionNode
    participant TE as ToolExecutionNode
    participant T as Tool

    LLM->>TD: Analyze query
    TD->>TD: Select tools
    TD->>TE: Tool decisions
    TE->>T: execute(**args)
    T-->>TE: Result
    TE-->>LLM: Tool results

Best Practices¶

1. Clear Naming¶

# Good: Descriptive name
def get_name(self) -> str:
    return "get_weather_forecast"

# Bad: Vague name
def get_name(self) -> str:
    return "tool1"

2. Helpful Descriptions¶

# Good: Detailed description
def get_description(self) -> str:
    return (
        "Get weather forecast for a city. "
        "Returns temperature, conditions, and humidity. "
        "Use for weather-related questions."
    )

# Bad: Minimal description
def get_description(self) -> str:
    return "Weather"

3. Comprehensive Schema¶

# Good: Detailed schema with descriptions
{
    "type": "object",
    "properties": {
        "city": {
            "type": "string",
            "description": "City name (e.g., 'London, UK')",
            "minLength": 2
        }
    },
    "required": ["city"]
}

4. Graceful Error Handling¶

def execute(self, **kwargs) -> dict:
    try:
        result = self.do_work(kwargs)
        return {"success": True, "data": result}
    except ValueError as e:
        return {"success": False, "error": f"Invalid input: {e}"}
    except ConnectionError as e:
        return {"success": False, "error": "Service unavailable"}
    except Exception as e:
        return {"success": False, "error": f"Unexpected error: {e}"}

5. Logging¶

import logging

logger = logging.getLogger(__name__)

def execute(self, **kwargs) -> dict:
    logger.info(f"Executing {self.get_name()} with {kwargs}")
    result = self.do_work(kwargs)
    logger.info(f"Result: {result}")
    return result