fit-boilerplate-wox/docs/tool-authoring-guide.md

Tool Authoring Guide

WxO acepta tools de tres formas. Acá te explico cuándo usar cuál y cómo.

Decisión rápida

Situación	Tipo
PoC / demo / mocks	Python @tool
Tu backend ya expone REST (FastAPI/Express/etc.)	OpenAPI
SaaS de terceros con MCP server	MCP
SaaS de terceros sin MCP	OpenAPI (escribís el spec) o Python wrapper

---

Tipo 1 — Python @tool (estilo Cotemar)

Cuándo

• Estás prototipando rápido
• Los sistemas externos son mocks separados
• La lógica del wrapper es simple
• Querés tener cada tool en su propia función testeable

Template

wxo/tools/python/_template_tools.py:

# === inline compat shim (NO eliminar — TRM lo necesita) ===
try:
    from ibm_watsonx_orchestrate.agent_builder.tools import tool
except ImportError:
    def tool(*args, **kwargs):
        def deco(f): return f
        return deco if not args else deco(args[0])
# ============================================================

import os
import requests
from pydantic import BaseModel, field_validator

from wxo.tools.python._observable_tool import observable_tool
from wxo.tools.python._coercion_helpers import _coerce_int, _coerce_list

BASE_URL = os.environ.get("BASE_URL", "")

class ResetPasswordInput(BaseModel):
    username: str
    notify: bool = True

@observable_tool(name="reset_password", domain="ad")
def reset_password(username: str, notify: bool = True) -> dict:
    """Reset AD password and return temp credentials."""
    resp = requests.post(
        f"{BASE_URL}/users/{username}/reset-password",
        json={"notify": notify},
        timeout=10
    )
    resp.raise_for_status()
    return resp.json()

Import

orchestrate tools import -k python \
  -f wxo/tools/python/ad_tools.py \
  -r wxo/tools/python/requirements.txt \
  --app-id ad_demo

Gotchas

• No usar imports relativos (from .._compat import ...). Inlinear el shim. (Issue I-010)
• Coerción de tipos siempre. El LLM stringifica todo. (Issue I-003)
• BASE_URL desde env, nunca hardcoded.

---

Tipo 2 — OpenAPI (estilo Dun)

Cuándo

• Tu backend ya define endpoints REST con tipos
• Querés single-source-of-truth (el spec)
• Tenés FastAPI/Express/etc. y solo necesitás exponer un subset a WxO

Template — endpoint público filtrado

wxo/tools/openapi/_backend_filter_endpoint.py:

from copy import deepcopy
from fastapi import FastAPI

# Allowlist de operaciones que WxO puede ver
PUBLIC_TOOLS = {
    "POST /api/v1/cases/run",
    "GET /api/v1/cases/{case_id}",
    "POST /api/v1/cases/{case_id}/approve",
}

def build_public_spec(app: FastAPI) -> dict:
    """Filtra el openapi.json de tu app a la allowlist PUBLIC_TOOLS,
    fuerza description per-operation, y resuelve $ref transitivamente."""
    full = app.openapi()
    spec = {
        "openapi": full["openapi"],
        "info": {**full["info"], "title": full["info"]["title"] + " (orchestrate)"},
        "servers": full.get("servers", [{"url": "https://CHANGEME/api/v1"}]),
        "paths": {},
        "components": {"schemas": {}},
        "security": full.get("security", []),
    }

    # Filtrar paths
    for path, item in full["paths"].items():
        for method, op in item.items():
            if method.upper() in ("GET", "POST", "PATCH", "PUT", "DELETE"):
                key = f"{method.upper()} {path}"
                if key in PUBLIC_TOOLS:
                    spec["paths"].setdefault(path, {})[method] = op
                    # description fallback (issue I-005)
                    op["description"] = op.get("description") or op.get("summary") or f"{method} {path}"
                    # security per-operation (issue I-006)
                    op["security"] = spec["security"]

    # Resolver $ref transitivamente
    needed_schemas = set()
    def walk(o):
        if isinstance(o, dict):
            for k, v in o.items():
                if k == "$ref" and isinstance(v, str) and v.startswith("#/components/schemas/"):
                    needed_schemas.add(v.split("/")[-1])
                else:
                    walk(v)
        elif isinstance(o, list):
            for x in o: walk(x)
    walk(spec["paths"])
    all_schemas = full.get("components", {}).get("schemas", {})
    # Resolver hasta closure
    pending = list(needed_schemas)
    while pending:
        s = pending.pop()
        if s not in all_schemas: continue
        spec["components"]["schemas"][s] = all_schemas[s]
        # Encontrar refs nuevos dentro
        before = set(needed_schemas)
        walk(all_schemas[s])
        for new in needed_schemas - before:
            pending.append(new)

    return spec

# Endpoint que sirve el spec filtrado
@app.get("/api/v1/orchestrate-tools-spec.json")
def public_spec():
    return build_public_spec(app)

Patch del servers[0].url en deploy time

scripts/deploy-wxo.sh baja el spec del backend live y reemplaza el server URL:

curl -s "$BACKEND_URL/api/v1/orchestrate-tools-spec.json" \
  | jq --arg url "$BACKEND_URL/api/v1" '.servers = [{"url": $url}]' \
  > /tmp/spec.json

orchestrate tools import -k openapi -f /tmp/spec.json --app-id mi_backend

Schema Pydantic con coerción

En tu backend, cada endpoint input model lleva field_validator:

from pydantic import BaseModel, field_validator
from wxo.tools.python._coercion_helpers import _coerce_int, _coerce_list

class RunCaseInput(BaseModel):
    case_id: str
    target_companyid: int
    profile_ids: list[str]
    _coerce_companyid = field_validator("target_companyid", mode="before")(_coerce_int)
    _coerce_profiles = field_validator("profile_ids", mode="before")(_coerce_list)

Observabilidad con write_audit

En cada handler:

from app.audit import write_audit

@router.post("/cases/run")
def run_case(input: RunCaseInput, x_orchestrate_token: str = Header(...)):
    write_audit(input.case_id, "tool.run_case.started", input.dict())
    result = do_the_work(input)
    write_audit(input.case_id, "tool.run_case.completed", result)
    return result

---

Tipo 3 — MCP

Cuándo

• El SaaS ya expone un MCP server (HubSpot, Outline, GitHub vía MCP, etc.)
• Querés que ADK descubra las tools automáticamente
• Necesitás permisos del sistema externo respetados

Template

wxo/tools/mcp/_template_mcp_connection.yaml:

spec_version: v1
kind: connection
name: hubspot_mcp
display_name: "HubSpot via MCP"
schema_version: "1.0"
auth_type: mcp
identifier: hubspot_mcp_conn
preference:
  - environment: draft
    schema_id: mcp
  - environment: live
    schema_id: mcp

Setup

orchestrate connections add -a hubspot_mcp
for ENV in draft live; do
  orchestrate connections configure -a hubspot_mcp --env $ENV --type team --kind mcp
  orchestrate connections set-credentials -a hubspot_mcp --env $ENV \
    -e "MCP_SERVER_URL=https://hubspot-mcp.example.com" \
    -e "MCP_TOKEN=$HUBSPOT_TOKEN"
done

ADK descubre las tools del server y las hace disponibles al agente. En el YAML del agente listás las que necesita:

tools:
  - hubspot_mcp/get_contact
  - hubspot_mcp/create_deal

Caveats

• La latencia depende del MCP server. Algunos son lentos.
• Los nombres de tools vienen del MCP server, no los podés renombrar.
• Coordiná con el equipo del MCP server qué tools va a exponer.

---

Híbrido

Un agente puede usar las tres formas al mismo tiempo:

tools:
  # MCP (descubiertas del server)
  - hubspot_mcp/get_contact
  - hubspot_mcp/update_deal
  # OpenAPI (importadas del backend propio)
  - run_full_case
  - get_case_status
  # Python (wrappers a mocks o sistemas pequeños)
  - lookup_internal_user

Lo único que comparten es el contrato observable: cada call emite la misma traza JSON.

---

Patrón "meta-tool" (issue I-001)

Si tu flujo es lineal (caso → A → B → C → D), colapsá los 4 tools en uno:

# Mal — el agente puede explotar con recursion_limit
tools:
  - get_company
  - search_credits
  - build_profile
  - package_zip
  - upload

# Bien — agente invoca 1 tool, backend orquesta el resto
tools:
  - run_full_case

El backend ejecuta los 4 pasos internamente y emite write_audit por cada uno. La UI reconstruye el timeline igual. El agente es feliz y no muere por recursion.

---

Checklist antes de mergear un tool nuevo

• ¿Usa @observable_tool, no @tool directo?
• ¿Tiene description (docstring) clara que el LLM va a leer?
• ¿El input model tiene @field_validator(mode="before") para todo int/list?
• ¿Lee BASE_URL del env, no hardcoded?
• ¿Si es OpenAPI, tiene description y security per-operation?
• ¿El agente que lo usa tiene <10 tools después de agregarlo?
• ¿Hay un scenario de eval que cubra la happy path?
• ¿./evals/lint_wxo_yaml.py pasa?