spoon-tool-development

Original source / Raw SKILL.md

---
name: spoon-tool-development
description: Develop tools for SpoonOS agents. Use when creating custom tools, MCP servers, toolkit extensions, or configuring tool managers.
---

# Tool Development

Build tools for SpoonOS agents using BaseTool and FastMCP.

## Tool Types

| Type | Use Case |
|------|----------|
| `BaseTool` | Local Python tools |
| `MCPTool` | External MCP servers |
| `FastMCP` | Build MCP servers |

## Quick Start

```python
from spoon_ai.tools.base import BaseTool
from pydantic import Field

class MyTool(BaseTool):
    name: str = "my_tool"
    description: str = "Tool description"
    parameters: dict = Field(default={
        "type": "object",
        "properties": {
            "param": {"type": "string"}
        },
        "required": ["param"]
    })

    async def execute(self, param: str) -> str:
        return f"Result: {param}"
```

## Scripts

| Script | Purpose |
|--------|---------|
| [base_tool.py](scripts/base_tool.py) | BaseTool implementation |
| [mcp_tool.py](scripts/mcp_tool.py) | MCPTool configuration |
| [fastmcp_server.py](scripts/fastmcp_server.py) | FastMCP server |
| [tool_manager.py](scripts/tool_manager.py) | ToolManager usage |

## References

| Reference | Content |
|-----------|---------|
| [error_handling.md](references/error_handling.md) | Error patterns |
| [testing.md](references/testing.md) | Testing tools |

## ToolManager Methods

| Method | Description |
|--------|-------------|
| `add_tool(tool)` | Add single tool |
| `remove_tool(name)` | Remove by name |
| `execute(name, input)` | Execute tool |
| `to_params()` | Get OpenAI specs |

## Best Practices

1. Use async for all I/O operations
2. Return ToolResult/ToolFailure
3. Validate inputs before execution
4. Implement proper error handling
5. Use environment variables for secrets


---

## Referenced Files

> The following files are referenced in this skill and included for context.

### scripts/base_tool.py

```python
#!/usr/bin/env python3
"""BaseTool implementation examples."""

import aiohttp
from typing import Optional, Any
from spoon_ai.tools.base import BaseTool, ToolResult, ToolFailure
from pydantic import Field


class CoinGeckoTool(BaseTool):
    """Tool for fetching cryptocurrency prices from CoinGecko."""

    name: str = "get_crypto_price"
    description: str = "Get current cryptocurrency price from CoinGecko API"
    parameters: dict = Field(default={
        "type": "object",
        "properties": {
            "coin_id": {
                "type": "string",
                "description": "CoinGecko coin ID (e.g., 'bitcoin', 'ethereum')"
            },
            "currency": {
                "type": "string",
                "description": "Target currency (e.g., 'usd', 'eur')",
                "default": "usd"
            }
        },
        "required": ["coin_id"]
    })

    base_url: str = "https://api.coingecko.com/api/v3"
    timeout: int = 30

    async def execute(self, coin_id: str, currency: str = "usd") -> str:
        url = f"{self.base_url}/simple/price"
        params = {
            "ids": coin_id,
            "vs_currencies": currency,
            "include_24hr_change": "true"
        }

        try:
            async with aiohttp.ClientSession() as session:
                async with session.get(
                    url,
                    params=params,
                    timeout=aiohttp.ClientTimeout(total=self.timeout)
                ) as resp:
                    if resp.status != 200:
                        return f"Error: API returned status {resp.status}"

                    data = await resp.json()

                    if coin_id not in data:
                        return f"Error: Coin '{coin_id}' not found"

                    price = data[coin_id][currency]
                    change = data[coin_id].get(f"{currency}_24h_change", 0)

                    return (
                        f"{coin_id.upper()} Price: ${price:,.2f} {currency.upper()}\n"
                        f"24h Change: {change:+.2f}%"
                    )

        except aiohttp.ClientError as e:
            return f"Network error: {str(e)}"
        except Exception as e:
            return f"Error: {str(e)}"


class SafeTool(BaseTool):
    """Tool with comprehensive error handling."""

    name: str = "safe_operation"
    description: str = "Safely execute operations with validation"
    parameters: dict = Field(default={
        "type": "object",
        "properties": {
            "data": {"type": "string"}
        },
        "required": ["data"]
    })

    async def execute(self, **kwargs) -> Any:
        # Input validation
        error = self._validate_inputs(**kwargs)
        if error:
            return ToolFailure(error)

        try:
            result = await self._do_operation(**kwargs)
            return ToolResult(output=result)
        except ValueError as e:
            return ToolFailure(f"Invalid input: {e}")
        except ConnectionError as e:
            return ToolFailure(f"Connection failed: {e}")
        except Exception as e:
            return ToolFailure(f"Unexpected error: {e}")

    def _validate_inputs(self, **kwargs) -> Optional[str]:
        required = self.parameters.get("required", [])
        for field in required:
            if field not in kwargs:
                return f"Missing required field: {field}"
        return None

    async def _do_operation(self, **kwargs) -> Any:
        return f"Processed: {kwargs.get('data')}"


async def main():
    # Test CoinGecko tool
    price_tool = CoinGeckoTool()
    result = await price_tool.execute(coin_id="bitcoin")
    print(result)


if __name__ == "__main__":
    import asyncio
    asyncio.run(main())

```

### scripts/fastmcp_server.py

```python
#!/usr/bin/env python3
"""FastMCP server example."""

from fastmcp import FastMCP
from pydantic import BaseModel, Field


# Create MCP server
mcp = FastMCP(name="MyToolsServer")


# Define input schemas
class PriceInput(BaseModel):
    symbol: str = Field(..., description="Token symbol (e.g., BTC, ETH)")
    currency: str = Field("usd", description="Target currency")


class AnalysisInput(BaseModel):
    data: str = Field(..., description="Data to analyze")
    format: str = Field("json", pattern="^(json|csv|text)$")


@mcp.tool()
async def get_price(input: PriceInput) -> str:
    """Get cryptocurrency price."""
    # Replace with actual implementation
    return f"Price of {input.symbol}: $50000 {input.currency.upper()}"


@mcp.tool()
async def analyze_data(input: AnalysisInput) -> str:
    """Analyze data in various formats."""
    return f"Analysis result for {input.format} data"


@mcp.resource("config://settings")
async def get_settings() -> str:
    """Get server configuration."""
    return '{"version": "1.0.0", "supported_chains": ["ethereum", "polygon"]}'


if __name__ == "__main__":
    mcp.run()

```