Project Builder

# Build Patterns — Detailed Reference

## Pattern A: Scheduled Task (Cron Job)

**Use when:** User wants recurring monitoring, alerts, or reports.

**Build order (each step verified before the next):**

**Step 1 — Data fetching (verify: real data comes back)**

**Preferred: Use `core.skill_tools` to call skill functions directly:**
```python
# skill_tools auto-discovers skills/*/exports.py — no manual path setup needed
from core.skill_tools import coingecko, coinglass

# Same function names as agent tools, namespaced by skill
prices = coingecko.coin_price(coin_ids=["bitcoin", "ethereum"], timestamps=["now"])
btc_price = prices[0]['price']

fr = coinglass.funding_rate(symbol="BTC")
binance_rate = [e for e in fr['exchanges_data'] if e['exchangeName'] == 'Binance'][0]['rate']
```

**Fallback: Direct API call (for skills without exports.py):**
```python
import requests, os, json

def fetch_prices():
    resp = requests.get("https://api.twelvedata.com/price", params={
        "symbol": "CL1,BZ1",
        "apikey": os.environ.get("TWELVEDATA_API_KEY", "")
    })
    return resp.json()

if __name__ == "__main__":
    result = fetch_prices()
    print(json.dumps(result, indent=2))
```
Run it: `python3 tasks/{id}/run.py`
✅ Pass = real prices printed, recent timestamps
❌ Fail = fix before proceeding

**Step 2 — Data validation (verify: bad data gets caught)**
```python
def validate_price(price_str, symbol):
    """Returns (float, error). Error is None if valid."""
    try:
        price = float(price_str)
    except (ValueError, TypeError):
        return None, f"{symbol}: invalid price '{price_str}'"
    if price <= 0 or price > 500:  # reasonable range for crude oil
        return None, f"{symbol}: price {price} out of range"
    return price, None
```
Test with both good and bad inputs.

**Step 3 — Logic/analysis (verify: output format is correct)**
```python
def analyze(prices, thresholds):
    alerts = []
    for symbol, data in prices.items():
        change_pct = data["change_pct"]
        if abs(change_pct) > thresholds["alert_pct"]:
            alerts.append(f"⚠️ {symbol}: {change_pct:+.1f}%")
    return alerts
```
Test with edge cases: exactly at threshold, just above, just below.

**Step 4 — Output formatting (verify: readable, correct numbers)**
```python
# Print only when there's something actionable
output = format_report(prices, alerts)
if output.strip():
    print(output)  # Non-empty stdout → auto-pushed to user
# Empty stdout → silent, no push, no cost
```
Run full pipeline, manually verify every number in output matches raw data.

**Step 5 — Register and activate**
```python
register_task(title="WTI Monitor", schedule="0 * * * *")
# Write verified run.py to tasks/{id}/
# One final dry-run
bash("python3 tasks/{id}/run.py")
# Only then:
activate_task(job_id)
```

### Using Skill Functions in Task Scripts (`core.skill_tools`)

Skills with `exports.py` expose their tool functions for direct use in task scripts.
No manual path setup needed — `core.skill_tools` auto-discovers all exports.

```python
from core.skill_tools import coingecko, coinglass, birdeye

# Same names as agent tools, namespaced by skill
prices = coingecko.coin_price(coin_ids=["bitcoin", "ethereum"], timestamps=["now"])
btc_price = prices[0]['price']

fr = coinglass.funding_rate(symbol="BTC")
binance = [e for e in fr['exchanges_data'] if e['exchangeName'] == 'Binance']
btc_funding = binance[0]['rate'] if binance else None

etf = coinglass.cg_btc_etf_flows()
```

**How to know what's available:**
- Tool names = SKILL.md frontmatter `tools:` list
- Import pattern: `from core.skill_tools import {skill_name}` → `{skill_name}.{tool_name}(...)`
- Functions are sync (safe for task scripts). Return types match the underlying API response.

**When skill_tools is NOT available:**
- Skills without `exports.py` (e.g. hyperliquid — needs wallet context)
- Async-only tools (twelvedata uses aiohttp)
- For these, use direct API calls via `requests` / `core.http_client`