Agent Tools¶

Tools give agents the ability to take actions — search databases, call APIs, run code, or interact with external services. amsdal_ml provides three built-in tool types.

PythonTool¶

Wraps any Python function (sync or async) as a tool. The agent sees the function's name, docstring, and parameter schema.

from amsdal_ml.agents.python_tool import PythonTool

async def get_customer_orders(customer_id: str, limit: int = 10) -> str:
    """Fetch recent orders for a customer.

    Args:
        customer_id: The customer's ID.
        limit: Maximum number of orders to return.
    """
    orders = await Order.objects.filter(customer_id=customer_id)[:limit]
    return str([o.to_dict() for o in orders])

tool = PythonTool(
    get_customer_orders,
    name='get_customer_orders',
    description='Fetch recent orders for a customer.',
)

The tool schema is automatically generated from the function signature. Pydantic FieldInfo is supported for extra validation.

How It Works¶

name and description are required constructor arguments
Parameters are extracted from the function signature and converted to JSON schema
Both sync and async functions are supported (sync functions run on the event loop directly)

RetrieverTool¶

A pre-built tool that exposes semantic search to agents:

from amsdal_ml.agents.retriever_tool import retriever_search

# Use directly as a PythonTool
tool = PythonTool(retriever_search, name='search', description='Semantic search in knowledge base.')

The retriever_search function accepts:

Parameter	Type	Default	Description
`query`	`str`	—	Search query
`k`	`int`	`5`	Number of results
`include_tags`	`Optional[list[str]]`	`None`	Include only these tags
`exclude_tags`	`Optional[list[str]]`	`None`	Exclude these tags

It uses OpenAIRetriever under the hood (lazily initialized).

ToolClient (MCP Client Tools)¶

Connect to an external MCP server and expose its tools to your agent:

from amsdal_ml.mcp_client.stdio_client import StdioClient

# Connect to an MCP server running as a subprocess
client = StdioClient(
    alias='my-server',
    module_or_cmd='my_mcp_server.main',
)

# List available tools
tools = await client.list_tools()

# Use with an agent
agent = FunctionalCallingAgent(
    model=OpenAIModel(),
    tools=[PythonTool(search_fn, name='search', description='Search.'), client],  # mix local and remote tools
)

Stdio Client¶

Launches an MCP server as a subprocess and communicates via stdio:

from amsdal_ml.mcp_client.stdio_client import StdioClient

client = StdioClient(
    alias='retriever',
    module_or_cmd='amsdal_ml.mcp_server.server_retriever_stdio',
)

Features: - Persistent or ephemeral sessions - Tool caching for efficiency - Automatic session recovery on errors - Configurable timeout

HTTP Client¶

Connects to an MCP server running over HTTP/SSE:

from amsdal_ml.mcp_client.http_client import HttpClient

client = HttpClient(
    alias='remote',
    url='http://localhost:8000/mcp/sse',
    headers={'Authorization': 'Bearer token...'},
)

Qualified Tool Names¶

When agents use tools from multiple clients, tool names are qualified with the client alias to avoid conflicts:

retriever.search        # from the 'retriever' client
remote.list_models      # from the 'remote' client
search_customers        # local PythonTool (no prefix)

Creating Custom Tools¶

Any async or sync function can become a tool via PythonTool:

import httpx
from amsdal_ml.agents.python_tool import PythonTool

async def check_weather(city: str) -> str:
    """Get current weather for a city."""
    async with httpx.AsyncClient() as client:
        resp = await client.get(f'https://wttr.in/{city}?format=3')
        return resp.text

weather_tool = PythonTool(check_weather, name='check_weather', description='Get current weather for a city.')

For more complex tools, implement the ToolClient protocol:

from amsdal_ml.mcp_client.base import ToolClient, ToolInfo

class MyToolClient:
    alias: str = 'my-client'

    async def list_tools(self) -> list[ToolInfo]:
        return [ToolInfo(alias=self.alias, name='my_tool', description='...', input_schema={...})]

    async def call(self, tool_name: str, args: dict[str, Any], *, timeout: float | None = None) -> Any:
        # Execute the tool
        ...