1. Introduction

Imagine you’re assembling a super-intelligent robot Agent. This robot needs a variety of tools to help you complete the task – just like Doraemon’s four-dimensional pocket. This article will teach you how to create these powerful tools to make your AI housekeeper more capable and efficient.

2. Two core tool design patterns

2.1 Synchronization Tool: Instant Response Mode

Imagine using a self-service coffee machine:

1. Insert a coin and press the “Americano” button
2. Wait a few seconds
3. The coffee flows out and is ready to be drunk

This is a typical synchronization tool pattern. The agent invokes the tool and waits for immediate results – quickly and easily.

class WeatherTool(BaseTool):
    ""Weather Finder - Sync Mode"""
    async def execute(self, city: str) -> dict:
        # Simple and straightforward, like pressing a coffee maker button
        weather_data = await self.weather_api.get_current(city)
        return {
            "status": "success",
            "data": {
                "temperature": weather_data.temp,
                "humidity": weather_data.humidity,
                "description": weather_data.desc
            }
        }

Usage Scenarios:

• Quick query: weather, exchange rates, simple calculations
• Simple operation: send messages, switch control
• Real-time feedback: CAPTCHA check, balance inquiry

2.2 Asynchronous Tools: Task Tracking Mode

Imagine ordering food through a food delivery app:

1. After placing an order, the APP gives you an order number
2. You can check the status of your order at any time
3. The APP will notify you when the order is completed

This is how asynchronous tools work, and are suitable for tasks that take a long time.

class DocumentAnalysisTool(BaseTool):
    Document Analysis Tool - Asynchronous Mode

async def start_task(self, file_path: str) -> str:
        # It's like placing a takeout order and returning the task ID
        task_id = str(uuid.uuid4())
        await self.task_queue.put({
            "task_id": task_id,
            "file_path": file_path,
            "status": "processing"
        })
        return task_id

async def get_status(self, task_id: str) -> dict:
        # It's like checking the status of a takeaway order
        task = await self.task_store.get(task_id)
        return {
            "task_id": task_id,
            "status": task["status"],
            "progress": task.get("progress", 0),
            "result": task.get("result", None)
        }

Usage Scenarios:

• Time-consuming operations: large file processing and data analysis
• Multi-step tasks: video rendering, report generation
• Progress tracking is required: model training, batch processing

3. Standardization of tool interfaces: Establish common specifications

Just as all appliances follow a uniform socket standard, our tool interfaces need to be standardized. This ensures that all tools work perfectly with the Agent.

3.1 Tool Description Specification

Imagine writing a product manual where you need to clearly tell the user:

• Functions of the tool
• What parameters are required
• What results will be returned

from pydantic import BaseModel, Field

class ToolSchema(BaseModel):
    Tool manual template
    name: str = Field(..., description="Tool Name")
    description: str = Field(..., description="Tool Description")
    parameters: dict = Field(..., description="Required parameters")
    required: list[str] = Field(default_factory=list, description="Required Parameters")

class Config:
        schema_extra = {
            "example": {
                "name": "Weather Query",
                "description": "Query the weather information of a specified city",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "city": {
                            "type": "string",
                            "description": "City Name"
                        }
                    }
                },
                "required": ["city"]
            }
        }

3.2 Unified Base Classes

Just as all appliances require power switches and power interfaces, all tools need to follow basic specifications:

class BaseTool(ABC):
    Basic Templates for All Tools

@abstractmethod
    def get_schema(self) -> ToolSchema:
        ""Tool Manual"""
        pass

def validate_input(self, params: Dict) -> Dict:
        """"Parameter checking, like a fuse in an appliance""""
        return ToolSchema(**params).dict()

@abstractmethod
    async def execute(self, **kwargs) -> Dict:
        ""Actual Function Execution""
        pass

4. Error handling: Make the tool more reliable

Just as household appliances need protection against water, shock, and overload, tools need comprehensive protection mechanisms.

4.1 Misclassification and Handling

Imagine handling a courier:

• Wrong address → The parameter is incorrect
• The System Maintenance → service is temporarily unavailable

The

• courier is too busy → needs to throttle and retries

class ToolError(Exception):
    ""Tool Error Base Class""
    def __init__(self, message: str, error_code: str, retry_after: Optional[int] = None):
        self.message = message
        self.error_code = error_code
        self.retry_after = retry_after

@error_handler
async def execute(self, **kwargs):
    try:
        # Perform specific actions
        result = await self._do_work(**kwargs)
        return {"status": "success", "data": result}
    except ValidationError:
        # The parameter is wrong, just like the wrong address
        return {"status": "error", "code": "INVALID_PARAMS"}
    except RateLimitError as e:
        # Need to limit the flow, just like the courier is too busy
        return {
            "status": "error", 
            "code": "RATE_LIMIT",
            "retry_after": e.retry_after
        }

4.2 Retry Mechanism

It’s like automatically scheduling a second delivery after the first delivery fails:

class RetryableTool(BaseTool):
    @retry(
        stop=stop_after_attempt(3), # Retry up to 3 times
        wait=wait_exponential(multiplier=1, min=4, max=10) # Increase the wait time
    )
    async def execute_with_retry(self, **kwargs):
        return await self.execute(**kwargs)

5. Performance optimization: Make the tool more efficient

5.1 Caching Mechanisms

It’s like a convenience store that puts hot items in a prominent position:

class CachedSearchTool(BaseTool):
    def __init__(self):
        self.cache = {} # Simple memory cache
        self.cache_ttl = 3600 # Cached for 1 hour

async def execute(self, query: str):
        # First check if it's on the "shelf".
        cache_key = f"search:{query}"
        if cache_key in self.cache:
            return self.cache[cache_key]

# If not, get it from the repository
        result = await self._do_search(query)
        self.cache[cache_key] = result
        return result

5.2 Concurrency Control

It’s like a hospital’s appointment system, controlling the number of simultaneous services:

class RateLimiter:
    def __init__(self, max_concurrent: int = 5):
        self._semaphore = Semaphore(max_concurrent) # Handle up to 5 requests at the same time

@asynccontextmanager
    async def acquire(self):
        async with self._semaphore:
            yield

class ApiTool(BaseTool):
    def __init__(self):
        self.rate_limiter = RateLimiter(max_concurrent=5)

async def execute(self, **kwargs):
        async with self.rate_limiter.acquire():
            return await self._call_api(**kwargs)

6. Testing and documentation: Ensure tool reliability

6.1 Unit Testing

It’s like a quality check before a new product is launched:

class TestWeatherTool:
    @pytest.mark.asyncio
    async def test_normal_weather(self):
        ""Test Normal Weather Query""
        tool = WeatherTool()
        result = await tool.execute(city="Beijing")
        assert result["status"] == "success"
        assert "temperature" in result["data"]

@pytest.mark.asyncio
    async def test_invalid_city(self):
        """Test Invalid City Name""
        tool = WeatherTool()
        result = await tool.execute(city="NonexistentCity")
        assert result["status"] == "error"

6.2 Documentation Standards

It’s like writing a detailed and clear product manual:

class WeatherTool(BaseTool):
    """
    Weather lookup tool

Function: Query the real-time weather information of a specified city

Examples:
    ```

python
    tool = WeatherTool()
    result = await tool.execute(city="Beijing")
    print(f"Temperature: {result['data']['temperature']}°C")

```

Notes:
    1. The city name must be a valid Chinese city name
    2. A maximum of 10 queries per minute
    """

7. Summary

Developing a good agent tool is like building a perfect toolbox:

1. Reasonable categorization of tools – synchronous/asynchronous has its own purpose
2. Standardized interface – easy to manage in a unified manner
3. Protection Mechanism – Handle all kinds of exceptions
4. Pursuit of efficiency – caching when needed and throttling when necessary
5. Quality focus – thorough testing, clear documentation

Remember: good tools can make the agent do more with less, while bad tools will limit the agent’s performance in every link.