Gap Analysis → API Discovery → Tool Creation → Validation → Integration
     ↓              ↓               ↓              ↓            ↓
  Coverage      Web Search      devtu-create   devtu-fix    Git PR
  Report        + Docs          patterns       validation   Ready

undefined

---

API has multiple endpoints?
  ├─ YES → Multi-operation tool (single class, multiple JSON wrappers)
  └─ NO → Consider if more endpoints likely in future
       ├─ YES → Still use multi-operation (future-proof)
       └─ NO → Single-operation acceptable

from typing import Dict, Any
from tooluniverse.tool import BaseTool
from tooluniverse.tool_utils import register_tool
import requests
import os

@register_tool("[APIName]Tool")
class [APIName]Tool(BaseTool):
    """Tool for [API Name] - [brief description]."""

    BASE_URL = "[API base URL]"

    def __init__(self, tool_config):
        super().__init__(tool_config)
        self.parameter = tool_config.get("parameter", {})
        self.required = self.parameter.get("required", [])
        # For optional API keys
        self.api_key = os.environ.get("[API_KEY_NAME]", "")

    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 required parameter: operation"}

        # Route to handlers
        if operation == "operation1":
            return self._operation1(arguments)
        elif operation == "operation2":
            return self._operation2(arguments)
        else:
            return {"status": "error", "error": f"Unknown operation: {operation}"}

    def _operation1(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
        """Description of operation1."""
        # Validate required parameters
        param1 = arguments.get("param1")
        if not param1:
            return {"status": "error", "error": "Missing required parameter: param1"}

        try:
            # Build request
            headers = {}
            if self.api_key:
                headers["Authorization"] = f"Bearer {self.api_key}"

            # Make API call
            response = requests.get(
                f"{self.BASE_URL}/endpoint",
                params={"param1": param1},
                headers=headers,
                timeout=30
            )
            response.raise_for_status()

            # Parse response
            data = response.json()

            # Return with data wrapper
            return {
                "status": "success",
                "data": data.get("results", []),
                "metadata": {
                    "total": data.get("total", 0),
                    "source": "[API Name]"
                }
            }

        except requests.exceptions.Timeout:
            return {"status": "error", "error": "API timeout after 30 seconds"}
        except requests.exceptions.HTTPError as e:
            return {"status": "error", "error": f"HTTP {e.response.status_code}: {e.response.text[:200]}"}
        except Exception as e:
            return {"status": "error", "error": f"Unexpected error: {str(e)}"}

[
  {
    "name": "[APIName]_operation1",
    "class": "[APIName]Tool",
    "description": "[What it does]. Returns [data format]. [Input details]. Example: [usage example]. [Special notes].",
    "parameter": {
      "type": "object",
      "required": ["operation", "param1"],
      "properties": {
        "operation": {
          "const": "operation1",
          "description": "Operation identifier (fixed)"
        },
        "param1": {
          "type": "string",
          "description": "Description of param1 with format/constraints"
        }
      }
    },
    "return_schema": {
      "oneOf": [
        {
          "type": "object",
          "properties": {
            "data": {
              "type": "array",
              "items": {
                "type": "object",
                "properties": {
                  "id": {"type": "string"},
                  "name": {"type": "string"}
                }
              }
            },
            "metadata": {
              "type": "object",
              "properties": {
                "total": {"type": "integer"},
                "source": {"type": "string"}
              }
            }
          }
        },
        {
          "type": "object",
          "properties": {
            "error": {"type": "string"}
          },
          "required": ["error"]
        }
      ]
    },
    "test_examples": [
      {
        "operation": "operation1",
        "param1": "real_value_from_api_docs"
      }
    ]
  }
]

undefined

undefined

undefined

undefined

undefined

undefined

undefined

undefined

1. API docs show: GET /items → returns [{id: "ABC123", ...}]
2. Make request: curl https://api.example.com/items
3. Extract: id = "ABC123"
4. Verify: curl https://api.example.com/items/ABC123 → 200 OK
5. Use in test_examples: {"operation": "get_item", "item_id": "ABC123"}

TOOLS_CONFIGS = {
    # ... existing entries ...
    "[api_category]": os.path.join(current_dir, "data", "[api_name]_tools.json"),
}

import json

with open("src/tooluniverse/data/[api_name]_tools.json") as f:
    tools = json.load(f)

for tool in tools:
    schema = tool.get("return_schema", {})

    # Must have oneOf
    assert "oneOf" in schema, f"{tool['name']}: Missing oneOf in return_schema"

    # oneOf must have 2 schemas (success + error)
    assert len(schema["oneOf"]) == 2, f"{tool['name']}: oneOf must have 2 schemas"

    # Success schema must have 'data' field
    success_schema = schema["oneOf"][0]
    assert "properties" in success_schema, f"{tool['name']}: Missing properties in success schema"
    assert "data" in success_schema["properties"], f"{tool['name']}: Missing 'data' field in success schema"

    print(f"✅ {tool['name']}: Schema valid")

PLACEHOLDER_PATTERNS = [
    "test", "dummy", "placeholder", "example", "sample",
    "xxx", "temp", "fake", "mock", "your_"
]

for tool in tools:
    examples = tool.get("test_examples", [])

    for i, example in enumerate(examples):
        for key, value in example.items():
            if isinstance(value, str):
                value_lower = value.lower()
                if any(pattern in value_lower for pattern in PLACEHOLDER_PATTERNS):
                    print(f"❌ {tool['name']}: test_examples[{i}][{key}] contains placeholder: {value}")
                else:
                    print(f"✅ {tool['name']}: test_examples[{i}][{key}] appears real")

import sys
sys.path.insert(0, 'src')

undefined

undefined

**Handle failures**:
- **404 ERROR**: Invalid test example ID → find real ID
- **Schema Mismatch**: return_schema doesn't match response → fix schema
- **Timeout**: API slow/down → increase timeout or add retry
- **Parameter Error**: Wrong parameter names → verify with API docs

undefined

---

undefined

undefined

undefined

undefined

undefined

Testing [APIName] tools...
Total: X tests
Passed: X (100.0%)
Failed: 0
Schema invalid: 0

from tooluniverse import ToolUniverse

tu = ToolUniverse()
tu.load_tools()

undefined

undefined

undefined

---

undefined

undefined

Discovery → [API1, API2, API3] → Create All → Validate All → Single PR

Discovery → API1 → Create → Validate → Integrate
         → API2 → Create → Validate → Integrate

Discovery → Generate Report → [Manual Review] → Schedule Implementation

[Existing Tools] → Validation → Fix Issues → Re-validate → Report

def _submit_job(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
    """Submit job and poll for completion."""
    # Submit
    submit_resp = requests.post(
        f"{self.BASE_URL}/jobs",
        json={"data": arguments.get("data")},
        timeout=30
    )
    job_id = submit_resp.json().get("job_id")

    # Poll (with timeout)
    for attempt in range(60):  # 2 min max
        status_resp = requests.get(
            f"{self.BASE_URL}/jobs/{job_id}",
            timeout=30
        )
        result = status_resp.json()

        if result.get("status") == "completed":
            return {"status": "success", "data": result.get("results")}
        elif result.get("status") == "failed":
            return {"status": "error", "error": result.get("error")}

        time.sleep(2)

    return {"status": "error", "error": "Job timeout after 2 minutes"}

{
  "is_async": true,
  "poll_interval": 2,
  "max_wait_time": 120
}

undefined

**JSON Config**:
```json
{
  "parameter": {
    "properties": {
      "operation": {
        "const": "search_items",
        "description": "SOAP operation name (required)"
      }
    }
  }
}

def _list_all(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
    """Fetch all pages."""
    all_results = []
    page = 1

    while True:
        response = requests.get(
            f"{self.BASE_URL}/items",
            params={"page": page, "limit": 100},
            timeout=30
        )
        data = response.json()

        results = data.get("results", [])
        if not results:
            break

        all_results.extend(results)

        if len(results) < 100:  # Last page
            break

        page += 1

    return {
        "status": "success",
        "data": all_results,
        "metadata": {"total_pages": page, "total_items": len(all_results)}
    }

undefined

undefined