File Structure
Required Files:
src/tooluniverse/my_api_tool.py - Implementationsrc/tooluniverse/data/my_api_tools.json - Tool definitionstests/unit/test_my_api_tool.py - Testsexamples/my_api_examples.py - Usage examples
Auto-Generated (don't create manually):
src/tooluniverse/tools/MyAPI_*.py - Wrappers
Pattern 1: Multi-Operation Tool (Recommended)
Python Class:
```python
from typing import Dict, Any
from tooluniverse.tool import BaseTool
from tooluniverse.tool_utils import register_tool
import requests
@register_tool("MyAPITool")
class MyAPITool(BaseTool):
"""Tool for MyAPI database."""
BASE_URL = "https://api.example.com/v1"
def __init__(self, tool_config):
super().__init__(tool_config)
self.parameter = tool_config.get("parameter", {})
self.required = self.parameter.get("required", [])
def run(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
"""Route to operation handler."""
operation = arguments.get("operation")
if not operation:
return {"status": "error", "error": "Missing: operation"}
if operation == "list_items":
return self._list_items(arguments)
elif operation == "search":
return self._search(arguments)
else:
return {"status": "error", "error": f"Unknown: {operation}"}
def _list_items(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
"""List items with pagination."""
try:
params = {}
if "limit" in arguments:
params["limit"] = arguments["limit"]
response = requests.get(
f"{self.BASE_URL}/items",
params=params,
timeout=30
)
response.raise_for_status()
data = response.json()
return {
"status": "success",
"data": data.get("items", []),
"total": data.get("total", 0)
}
except requests.exceptions.Timeout:
return {"status": "error", "error": "Timeout after 30s"}
except requests.exceptions.HTTPError as e:
return {"status": "error", "error": f"HTTP {e.response.status_code}"}
except Exception as e:
return {"status": "error", "error": str(e)}
def _search(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
"""Search items by query."""
query = arguments.get("query")
if not query:
return {"status": "error", "error": "Missing: query"}
try:
response = requests.get(
f"{self.BASE_URL}/search",
params={"q": query},
timeout=30
)
response.raise_for_status()
data = response.json()
return {
"status": "success",
"results": data.get("results", []),
"count": data.get("count", 0)
}
except requests.exceptions.RequestException as e:
return {"status": "error", "error": f"API failed: {str(e)}"}
```
JSON Configuration:
```json
[
{
"name": "MyAPI_list_items",
"class": "MyAPITool",
"description": "List items from database with pagination. Returns item IDs and names. Supports filtering by status and type. Example: limit=10 returns first 10 items.",
"parameter": {
"type": "object",
"required": ["operation"],
"properties": {
"operation": {
"const": "list_items",
"description": "Operation type (fixed)"
},
"limit": {
"type": "integer",
"description": "Max results (1-100)",
"minimum": 1,
"maximum": 100
}
}
},
"return": {
"type": "object",
"properties": {
"status": {"type": "string", "enum": ["success", "error"]},
"data": {"type": "array"},
"total": {"type": "integer"},
"error": {"type": "string"}
},
"required": ["status"]
},
"test_examples": [
{
"operation": "list_items",
"limit": 10
}
]
}
]
```
Pattern 2: Async Polling (Job-Based APIs)
```python
import time
def _submit_job(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
"""Submit job and poll for results."""
try:
# Submit
submit_response = requests.post(
f"{self.BASE_URL}/jobs/submit",
json={"data": arguments.get("data")},
timeout=30
)
submit_response.raise_for_status()
job_id = submit_response.json().get("job_id")
# Poll
for attempt in range(60): # 2 min max
status_response = requests.get(
f"{self.BASE_URL}/jobs/{job_id}/status",
timeout=30
)
status_response.raise_for_status()
result = status_response.json()
if result.get("status") == "completed":
return {
"status": "success",
"data": result.get("results"),
"job_id": job_id
}
elif result.get("status") == "failed":
return {
"status": "error",
"error": result.get("error"),
"job_id": job_id
}
time.sleep(2) # Poll every 2s
return {"status": "error", "error": "Timeout after 2 min"}
except requests.exceptions.RequestException as e:
return {"status": "error", "error": str(e)}
```
JSON Best Practices
Tool Naming (β€55 chars for MCP):
- Template:
{API}_{action}_{target} - β
Good:
FDA_get_drug_info (20 chars) - β Bad:
FDA_get_detailed_drug_information_with_history (55+ chars)
Description (150-250 chars, high-context):
```json
{
"description": "Search database for items. Returns up to 100 results with scores. Supports wildcards (* ?) and Boolean operators (AND, OR, NOT). Example: 'protein AND membrane' finds membrane proteins."
}
```
Include: What it returns, data source, use case, input format, example
test_examples (MUST be real):
```json
{
"test_examples": [
{
"operation": "search",
"query": "protein", // β
Real, common term
"limit": 10
}
]
}
```
β Don't use: "id": "XXXXX", "placeholder": "example_123"
β
Do use: Real IDs from actual API documentation
---
|