Revisión de Antipatrones — AgentHub v0.5

🔴

Críticos

🟠

Altos

🟡

Medios

🔵

Bajos

✅

OK (14 chequeados)

Hallazgos detectados

Crítico API Key y credenciales hardcodeadas Infraestructura

La clave de API de Anthropic (sk-ant-api03-XXXX) y el host de base de datos de producción están escritos directamente en el código fuente. Cualquier commit expone estas credenciales en el historial de Git — un desastre de seguridad antes del primer cliente beta.

📄 agents/llm_caller.py — líneas 4-5

BAD — Actual

API_KEY = "sk-ant-api03-XXXX"
DB_HOST = "prod-agenthub.cluster.internal"

def call_claude(prompt):
    r = requests.post(
        url,
        headers={"x-api-key": API_KEY}
    )

GOOD — Corregido

from pydantic_settings import BaseSettings
from pydantic import Field

class Settings(BaseSettings):
    anthropic_api_key: str = Field(
        alias="ANTHROPIC_API_KEY"
    )
    db_host: str = Field(alias="DB_HOST")

settings = Settings()  # lee de .env

Acción: Crear .env (en .gitignore), usar pydantic-settings. Rotar la clave comprometida inmediatamente en console.anthropic.com.

Crítico Bare except silencioso — errores de LLM ocultos Manejo de Errores

El bloque except Exception: pass traga cualquier error silenciosamente. Si la API de Anthropic devuelve 429 (rate limit), 401 (clave inválida) o la red falla, la función retorna None sin aviso. Los agentes de los clientes fallarán sin logs ni alertas.

📄 agents/llm_caller.py — líneas 13-14

BAD — Actual

    except Exception:
        pass  # silencioso — bug oculto

GOOD — Corregido

    except httpx.TimeoutException as e:
        logger.warning("LLM timeout",
            prompt_len=len(prompt))
        raise AgentTimeoutError(str(e))
    except httpx.HTTPStatusError as e:
        logger.error("LLM API error",
            status=e.response.status_code)
        raise AgentAPIError(e.response.status_code)

Acción: Capturar excepciones específicas de httpx. Añadir structured logging con structlog. Nunca usar pass en un except de producción.

Alto Blocking I/O dentro de función async Recursos / Async

time.sleep(2) y requests.get() dentro de async def process_agents() bloquean el event loop de asyncio. Con 10 clientes beta ejecutando agentes en paralelo, esto puede congelar toda la instancia durante segundos.

📄 agents/llm_caller.py — líneas 17-22

BAD — Actual

async def process_agents(agents):
    time.sleep(2)     # bloquea el loop!
    results = []
    for agent in agents:
        result = call_claude(agent["prompt"])
        results.append(result)
    return results

GOOD — Corregido

async def process_agents(
    agents: list[AgentTask]
) -> list[AgentResult]:
    await asyncio.sleep(2)  # no bloquea
    tasks = [
        call_claude_async(a.prompt)
        for a in agents
    ]
    return await asyncio.gather(*tasks,
                                return_exceptions=True)

Acción: Migrar a httpx.AsyncClient. Usar asyncio.gather() para ejecutar agentes en paralelo — reducirá latencia x10 con 10 agentes simultáneos.

Alto Modelo ORM expuesto directamente en API Arquitectura

El endpoint GET /agents/{id} retorna el modelo SQLAlchemy directamente. Esto expone campos internos (timestamps, foreign keys, estados de base de datos), dificulta versionar la API y acopla la BD al contrato público.

📄 api/routes.py — línea 5

BAD — Actual

from db.models import AgentModel

@app.get("/agents/{id}")
def get_agent(id: str) -> AgentModel:
    agent = db.query(AgentModel).get(id)
    return agent  # expone internos

GOOD — Corregido

class AgentResponse(BaseModel):
    id: str
    name: str
    status: AgentStatus
    created_at: datetime

@app.get("/agents/{id}")
def get_agent(id: str) -> AgentResponse:
    agent = db.query(AgentModel).get(id)
    return AgentResponse.model_validate(
        agent, from_attributes=True
    )

Acción: Crear schemas Pydantic en api/schemas.py para cada endpoint. Añadir model_config = ConfigDict(from_attributes=True).

Alto Sin validación de input en create_agent Manejo de Errores

AgentModel(**data) acepta cualquier dict sin validar. Un cliente puede enviar campos extra, tipos incorrectos o campos obligatorios vacíos. Los errores aparecerán profundo en SQLAlchemy con mensajes confusos.

📄 api/routes.py — línea 10

BAD — Actual

def create_agent(data: dict):
    # sin validación — crashea profundo
    return AgentModel(**data)

GOOD — Corregido

class CreateAgentInput(BaseModel):
    name: str = Field(min_length=1, max_length=100)
    prompt: str = Field(min_length=10)
    model: str = "claude-3-5-sonnet-20241022"

def create_agent(data: dict) -> Agent:
    validated = CreateAgentInput.model_validate(data)
    return Agent.from_input(validated)

Acción: Validar en el borde de la API. Los errores de validación deben retornar HTTP 422 con mensaje claro, no 500.

Alto Over-mocking en tests — no verifica comportamiento real Testing

El test mockea db, llm y cache pero no verifica nada real. Podría pasar incluso si create_agent lanza una excepción internamente. El test da falsa seguridad.

📄 tests/test_agents.py — líneas 4-8

BAD — Actual

def test_create_agent():
    mock_db = Mock()
    mock_llm = Mock()
    mock_cache = Mock()
    # no testea nada real

GOOD — Corregido

def test_create_agent_success(db_session):
    agent = service.create_agent(valid_data, db_session)
    assert agent.id is not None
    assert agent.name == "Mi Agente"

def test_create_agent_invalid_prompt():
    with pytest.raises(ValidationError, match="min_length"):
        service.create_agent({"name": "x", "prompt": ""})

def test_create_agent_missing_name():
    with pytest.raises(ValidationError):
        service.create_agent({"prompt": "Hola"})

Acción: Usar pytest-asyncio + fixture de BD en memoria (SQLite). Testear paths de error y edge cases. Solo mockear APIs externas (Anthropic).

Medio Sin type hints en funciones públicas Tipado

call_claude(prompt) y process_agents(agents) no tienen type hints. El IDE no puede detectar errores en tiempo de desarrollo y mypy no puede hacer static analysis.

📄 agents/llm_caller.py — líneas 7, 17

BAD

def call_claude(prompt):
    ...
async def process_agents(agents):
    ...

GOOD

def call_claude(prompt: str) -> LLMResponse:
    ...
async def process_agents(
    agents: list[AgentTask]
) -> list[AgentResult]:
    ...

Acción: Añadir mypy al pipeline de CI. Definir tipos en agents/types.py.

Medio Sin retry centralizado para llamadas al LLM Infraestructura

La llamada a Anthropic no tiene retry. Las APIs de LLM son propensas a errores transitorios (rate limit 429, timeout puntual). Sin retry, el primer fallo del cliente en producción resultará en error visible.

📄 agents/llm_caller.py — línea 7

BAD

def call_claude(prompt: str):
    r = requests.post(url, ...)
    return r.json()

GOOD

from tenacity import retry, stop_after_attempt
from tenacity import wait_exponential

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(min=1, max=10),
    reraise=True
)
async def call_claude(prompt: str) -> LLMResponse:
    ...

Acción: Usar tenacity con backoff exponencial. Retry SOLO en el caller, no en capa de negocio (evitar double retry).

Medio Batch de agentes para en el primer error Manejo de Errores

Si uno de los 10 agentes en el batch falla, asyncio.gather() sin return_exceptions=True cancela todo el batch. Un fallo en agente #3 deja sin resultado a los agentes #4-#10.

📄 agents/llm_caller.py (diseño del gather)

BAD

results = await asyncio.gather(*tasks)
# para en primer error

GOOD

raw = await asyncio.gather(*tasks,
    return_exceptions=True)
succeeded = {i: r for i, r in enumerate(raw)
             if not isinstance(r, Exception)}
failed = {i: r for i, r in enumerate(raw)
          if isinstance(r, Exception)}
return BatchResult(succeeded, failed)

Acción: Usar return_exceptions=True y retornar BatchResult con partial failures. Los clientes deben ver qué agentes fallaron y cuáles no.

Checklist de cierre — Pre-release 0.5

❌ No scattered retry logic (centralizado)

✅ No double retry (app + infra)

❌ No hard-coded config o secrets

✅ No tipos internos expuestos (ORM)

❌ No tipos ORM en respuesta API

✅ Recursos cerrados con context managers

❌ No bare except Exception: pass

❌ No partial failures ignorados en batch

❌ Validación de input en boundaries

❌ No blocking calls en async

❌ Type hints en funciones públicas

❌ Colecciones con type params

❌ Paths de error testeados

❌ Edge cases cubiertos

Plan de acción — Antes del release 0.5

P	Acción	Archivo	Responsable	Estimación
P1	Rotar API key comprometida + migrar a pydantic-settings	`llm_caller.py`	Senior	1h
P1	Reemplazar `except Exception: pass` con manejo específico	`llm_caller.py`	Senior	2h
P2	Migrar a httpx async + asyncio.gather con return_exceptions	`llm_caller.py`	Junior A	3h
P2	Crear DTOs Pydantic en api/schemas.py	`api/routes.py`	Junior B	2h
P2	Añadir CreateAgentInput con validaciones Pydantic	`api/routes.py`	Junior A	1h
P3	Añadir @retry (tenacity) centralizado en llm_caller	`llm_caller.py`	Junior B	1h
P3	Reescribir tests con pytest-asyncio + tests de error paths	`test_agents.py`	Junior A	4h
P4	Añadir type hints completos + mypy a CI	Todos	Junior B	3h

Revisión de Antipatrones Python