# Capability: claude-direct Hablar directamente con `https://api.anthropic.com/v1/messages` usando el token OAuth de Claude Code (Claude Max), sin lanzar la CLI `claude` ni necesitar una API key de pago separada. 3 funciones Python en `domain: core`. ## Funciones | ID | Firma | Que hace | |---|---|---| | `load_claude_oauth_token_py_core` | `def load_claude_oauth_token(credentials_path: str = "", refresh_if_expired: bool = True) -> str` | Lee el access token OAuth desde `~/.claude/.credentials.json`. Verifica expiry (ms-epoch). Intenta refresh best-effort si expirado. | | `stream_anthropic_messages_py_core` | `def stream_anthropic_messages(messages: list, model: str = "claude-opus-4-8", ...) -> Iterator[dict]` | POST streaming a `/v1/messages`. Yield de eventos normalizados: `text`, `tool_use_start`, `tool_input_delta`, `done`, `error`. Parser SSE puro testeable por separado. | | `run_claude_tool_loop_py_core` | `def run_claude_tool_loop(messages, tools, dispatch, ...) -> dict` | Bucle agentico tool-use. Llama `stream_anthropic_messages` en loop, despacha tools via `dispatch{name: callable}`, anade `tool_result`, repite hasta `end_turn` o `max_iters`. | ## Ejemplo canonico end-to-end ### Pregunta simple (sin tools) ```python import sys sys.path.insert(0, "python/functions/core") from stream_anthropic_messages import stream_anthropic_messages text = "" for event in stream_anthropic_messages( messages=[{"role": "user", "content": "di solo PONG"}], model="claude-haiku-4-5-20251001", max_tokens=32, ): if event["type"] == "text": text += event["text"] print(event["text"], end="", flush=True) elif event["type"] == "done": print(f"\n[stop={event['stop_reason']}]") # Output: PONG # [stop=end_turn] ``` ### Bucle agentico con tool propia ```python import sys sys.path.insert(0, "python/functions/core") from run_claude_tool_loop import run_claude_tool_loop from datetime import datetime tools = [ { "name": "get_time", "description": "Devuelve la hora actual en formato HH:MM:SS.", "input_schema": {"type": "object", "properties": {}, "required": []}, } ] dispatch = { "get_time": lambda _inp: datetime.now().strftime("%H:%M:%S"), } result = run_claude_tool_loop( messages=[{"role": "user", "content": "que hora es exactamente ahora?"}], tools=tools, dispatch=dispatch, model="claude-haiku-4-5-20251001", on_text=lambda d: print(d, end="", flush=True), ) print(f"\n[iters={result['iterations']} stop={result['stop_reason']}]") # Claude llama a get_time() -> "14:32:07" # Luego responde: "Ahora son las 14:32:07." ``` ### Solo leer el token (para uso manual) ```python import sys sys.path.insert(0, "python/functions/core") from load_claude_oauth_token import load_claude_oauth_token token = load_claude_oauth_token(refresh_if_expired=False) # Pasar como header: {"authorization": f"Bearer {token}"} ``` ## Fronteras - **NO cubre** el flujo de refresh OAuth (endpoint no documentado publicamente) — el refresh es best-effort y puede fallar silenciosamente. - **NO es un cliente completo** de la API de Anthropic: solo `/v1/messages` con streaming. Files, embeddings, etc. quedan fuera. - **NO reemplaza** el uso de API keys oficiales para produccion — este grupo es exclusivamente para uso local del token OAuth de Claude Max. - **NO gestiona rate limits** — el caller debe manejar errores `{"type": "error"}` con `429` en el mensaje. ## Prerequisitos - Claude Code instalado y usuario logueado (`~/.claude/.credentials.json` debe existir). - `httpx` disponible en el venv: `python/.venv/bin/python3 -c "import httpx"`. - Token fresco (Claude Code normalmente lo renueva en background mientras esta abierto).