fn_registry/docs/capabilities/claude-direct.md

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)

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

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)

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).