← Volver al listado de tecnologías ← Índice de Bots de Telegram con Claude

Cap 5: Claude Code como Backend Local

1 de abril de 2026 Por: Artiko

claude-codesubprocesslocalclitelegramautomatizacion

La idea

Claude Code (claude) es la CLI de Anthropic que puede leer tu codebase, editar archivos y ejecutar comandos. Al conectarlo con Telegram, puedes:

Pedirle que haga un commit desde el móvil
Revisar el estado de tu proyecto mientras estás fuera
Ejecutar tareas de refactor o debugging remotamente

flowchart LR
    PHONE([Móvil\nTelegram]) -->|"arregla el bug en auth.py"| BOT[Bot Python\nlocal]
    BOT -->|subprocess| CC[claude --print\n--output-format json]
    CC -->|lee/edita| REPO[Tu repositorio\nlocal]
    CC -->|respuesta JSON| BOT
    BOT -->|resultado| PHONE

Requisitos

Claude Code instalado: npm install -g @anthropic-ai/claude-code
Bot corriendo en la misma máquina que el repositorio
Autenticado en Claude Code: claude --version debe funcionar

Claude Code en modo no-interactivo

El flag --print hace que Claude Code ejecute la tarea y retorne la respuesta sin interfaz interactiva. --output-format json da un output estructurado:

claude --print "¿Cuántos archivos .py hay en src/?" --output-format json

Output:

{
  "type": "result",
  "result": "Hay 12 archivos Python en src/.",
  "session_id": "abc123",
  "cost_usd": 0.0023,
  "duration_ms": 1847
}

Flags disponibles:

Flag	Descripción
`--print` / `-p`	Modo no-interactivo, imprime resultado y sale
`--output-format json`	Output en JSON estructurado
`--output-format stream-json`	Streaming de eventos JSON
`--max-turns N`	Límite de iteraciones del agente
`--system-prompt "..."`	System prompt adicional
`--allowedTools`	Herramientas permitidas
`--disallowedTools`	Herramientas bloqueadas

services/claude_code.py

import asyncio
import json
import subprocess
from pathlib import Path

# Directorio del repositorio que Claude Code puede modificar
REPO_PATH = Path.home() / "mi-proyecto"  # ajusta a tu ruta

# Herramientas permitidas (seguridad: no permitir Bash sin límite)
ALLOWED_TOOLS = [
    "Read",
    "Grep",
    "Glob",
    "Write",
    "Edit",
]

async def run_claude_code(prompt: str, allow_writes: bool = False) -> dict:
    """
    Ejecuta Claude Code con el prompt dado y retorna el resultado.
    
    allow_writes: si True, permite Write y Edit además de Read.
    """
    tools = ALLOWED_TOOLS if allow_writes else ["Read", "Grep", "Glob"]

    cmd = [
        "claude",
        "--print",
        "--output-format", "json",
        "--max-turns", "10",
        "--allowedTools", ",".join(tools),
        prompt,
    ]

    process = await asyncio.create_subprocess_exec(
        *cmd,
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.PIPE,
        cwd=str(REPO_PATH),
    )

    stdout, stderr = await asyncio.wait_for(
        process.communicate(),
        timeout=120,  # 2 minutos máximo
    )

    if process.returncode != 0:
        return {
            "error": stderr.decode().strip() or "Claude Code retornó error",
            "result": None,
        }

    try:
        data = json.loads(stdout.decode())
        return {"result": data.get("result", ""), "error": None}
    except json.JSONDecodeError:
        # Fallback: retornar el texto plano si no es JSON válido
        return {"result": stdout.decode().strip(), "error": None}

Comandos especializados en el bot

Agrega comandos dedicados para operaciones sobre el repo:

# handlers/commands.py (agregar)

from services.claude_code import run_claude_code

async def cmd_repo(update: Update, context: ContextTypes.DEFAULT_TYPE):
    """
    /repo <pregunta> — consulta de solo lectura sobre el codebase
    Ejemplo: /repo ¿Dónde está el handler de autenticación?
    """
    if not context.args:
        await update.message.reply_text("Uso: /repo <pregunta sobre el código>")
        return

    prompt = " ".join(context.args)
    chat_id = update.effective_chat.id

    stop = asyncio.Event()
    typing_task = asyncio.create_task(typing_loop(context.bot, chat_id, stop))

    try:
        resultado = await run_claude_code(prompt, allow_writes=False)
    finally:
        stop.set()
        typing_task.cancel()

    if resultado["error"]:
        await update.message.reply_text(f"Error: {resultado['error']}")
    else:
        await send_long_message(update, resultado["result"])

async def cmd_edit(update: Update, context: ContextTypes.DEFAULT_TYPE):
    """
    /edit <tarea> — permite que Claude Code modifique archivos.
    Solo disponible para usuarios autorizados.
    """
    if not context.args:
        await update.message.reply_text("Uso: /edit <descripción de la tarea>")
        return

    prompt = " ".join(context.args)
    chat_id = update.effective_chat.id

    # Confirmación antes de ejecutar cambios
    await update.message.reply_text(
        f"Voy a ejecutar esta tarea sobre el repositorio:\n\n"
        f"_{prompt}_\n\n"
        f"¿Confirmas? Responde /confirmar o /cancelar",
        parse_mode="Markdown"
    )
    # Guardamos la tarea pendiente en context.user_data
    context.user_data["tarea_pendiente"] = prompt

async def cmd_confirmar(update: Update, context: ContextTypes.DEFAULT_TYPE):
    prompt = context.user_data.pop("tarea_pendiente", None)
    if not prompt:
        await update.message.reply_text("No hay tarea pendiente.")
        return

    chat_id = update.effective_chat.id
    stop = asyncio.Event()
    typing_task = asyncio.create_task(typing_loop(context.bot, chat_id, stop))

    try:
        resultado = await run_claude_code(prompt, allow_writes=True)
    finally:
        stop.set()
        typing_task.cancel()

    if resultado["error"]:
        await update.message.reply_text(f"Error: {resultado['error']}")
    else:
        await send_long_message(update, resultado["result"])

Registrar los handlers en bot.py:

app.add_handler(CommandHandler("repo", cmd_repo))
app.add_handler(CommandHandler("edit", cmd_edit))
app.add_handler(CommandHandler("confirmar", cmd_confirmar))
app.add_handler(CommandHandler("cancelar", lambda u, c: u.message.reply_text("Cancelado.")))

Streaming con —output-format stream-json

Para tareas largas, el streaming muestra el progreso en tiempo real:

async def run_claude_code_streaming(
    prompt: str,
    update: Update,
    context,
    allow_writes: bool = False,
) -> str:
    """Ejecuta Claude Code con streaming y actualiza el mensaje de Telegram."""
    tools = ["Read", "Grep", "Glob"] if not allow_writes else ALLOWED_TOOLS

    cmd = [
        "claude", "--print",
        "--output-format", "stream-json",
        "--max-turns", "10",
        "--allowedTools", ",".join(tools),
        prompt,
    ]

    sent = await update.message.reply_text("Procesando...")
    resultado_final = ""

    process = await asyncio.create_subprocess_exec(
        *cmd,
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.PIPE,
        cwd=str(REPO_PATH),
    )

    async for line in process.stdout:
        if not line.strip():
            continue
        try:
            event = json.loads(line)
            tipo = event.get("type", "")

            if tipo == "assistant" and event.get("message", {}).get("content"):
                # Contenido de texto del asistente
                for bloque in event["message"]["content"]:
                    if bloque.get("type") == "text":
                        resultado_final += bloque["text"]
                        try:
                            await sent.edit_text(resultado_final[:4000])
                        except Exception:
                            pass

            elif tipo == "tool_use":
                # Mostrar qué herramienta está usando
                tool_name = event.get("tool_name", "herramienta")
                try:
                    await sent.edit_text(f"[Usando {tool_name}...]\n\n{resultado_final[:3900]}")
                except Exception:
                    pass

            elif tipo == "result":
                resultado_final = event.get("result", resultado_final)

        except json.JSONDecodeError:
            pass

    await process.wait()
    await sent.edit_text(resultado_final[:4000] or "Tarea completada.")
    return resultado_final

Consideraciones de seguridad

El acceso a Claude Code con permisos de escritura sobre tu repositorio es poderoso y potencialmente destructivo. Medidas recomendadas:

1. Restringir a un solo usuario

# .env
ALLOWED_USER_ID=123456789  # solo tú puedes usar /edit y /repo

2. Separar lectura de escritura

/repo — solo lectura, sin confirmación
/edit — escritura, requiere /confirmar

3. Limitar herramientas

# Nunca incluir Bash sin restricciones
SAFE_TOOLS = ["Read", "Grep", "Glob"]
WRITE_TOOLS = ["Read", "Grep", "Glob", "Write", "Edit"]
# Evitar: ["Bash"]  — riesgo de ejecución arbitraria

4. Timeout obligatorio

await asyncio.wait_for(process.communicate(), timeout=120)

5. Nunca exponer el bot públicamente — bot en modo polling, sin webhook, sin URL pública.

Ejemplo de sesión real

Tú → /repo ¿Dónde se valida el JWT?
Bot → En src/middleware/auth.py línea 34, función verify_token().
      Usa PyJWT con algoritmo HS256 y verifica expiración.

Tú → /repo ¿El token tiene refresh?
Bot → No hay refresh token implementado. Solo access token
      con expiración en 24h (config en settings.py línea 12).

Tú → /edit Agregar endpoint /auth/refresh que genere nuevo access token
Bot → Voy a ejecutar esta tarea sobre el repositorio:
      "Agregar endpoint /auth/refresh..."
      ¿Confirmas? Responde /confirmar o /cancelar

Tú → /confirmar
Bot → [Usando Read...]
      [Usando Write...]
      Agregué el endpoint /auth/refresh en src/routes/auth.py.
      Crea un nuevo access token si el existente es válido y
      expira en menos de 1 hora. Tests agregados en tests/test_auth.py.