Extender el runtime sin tocar el core: desarrollo de plugins con jg_agente_sdk

"El mayor síntoma de fragilidad en una arquitectura de software es que para dotar a un agente de una nueva habilidad tengas que abrir el repositorio del núcleo, modificar dependencias y volver a desplegar el motor de ejecución. Un runtime robusto debe ser hermético, ciego al dominio y radicalmente agnóstico. Las capacidades se extienden en la periferia a través de contratos tipados estrictos, paquetes desacoplados y herramientas de prueba que permitan verificar cualquier plugin en milisegundos sin necesidad de levantar bases de datos relacionales ni orquestadores pesados."

1. Resumen Ejecutivo: La Filosofía del Núcleo Hermético

En la construcción de plataformas corporativas de IA, la tentación más común es tratar el motor de ejecución del agente como un cajón de sastre: se incrustan consultas SQL específicas del negocio, llamadas a APIs propietarias o scripts de infraestructura directamente dentro del repositorio del core. A medida que la organización crece y surgen decenas de casos de uso distintos, el core se convierte en un monolito inmanejable con conflictos de dependencias, despliegues de alto riesgo y ciclos de desarrollo lentos.

Para resolver este acoplamiento estructural, hay que aplicar una separación tajante: el core (jg_agente) no contiene herramientas de negocio preinstaladas. Su única responsabilidad es ejecutar la máquina de estados, gobernar el acceso mediante SecurityGuard y persistir checkpoints. Todas las capacidades operativas (interacción con bases de datos, APIs de mensajería, pipelines de DevOps, análisis audiovisual) se implementan como plugins satélites totalmente independientes.

Para gobernar esta frontera se diseñó jg_agente_sdk: una biblioteca pública (licencia MIT) publicada en PyPI, ligera y con cero dependencias pesadas (apoyada exclusivamente en Pydantic v2), que define los contratos canónicos de herramientas y motores cognitivos. En este artículo detallamos la especificación de estos contratos, la generación rápida de plugins mediante la herramienta de línea de comandos jg, y el patrón de pruebas unitarias ultrarrápidas con MockPluginRunner que permite validar plugins en milisegundos sin dependencias externas.

2. Puntos de Fricción: El Acoplamiento de Herramientas en el Core

Al examinar plataformas de agentes que fracasan al intentar escalar en equipos multidisciplinares, se identifican con frecuencia tres errores estructurales:

1. Infierno de Dependencias (Dependency Hell): Si el plugin de DevOps requiere una versión concreta de boto3 o kubernetes, y el plugin de procesamiento audiovisual requiere dependencias pesadas de procesamiento multimedia, instalar todo en el entorno virtual del agente genera colisiones de paquetes y produce imágenes Docker de varios gigabytes.

2. Falta de Tipado y Contratos Laxos: En muchos frameworks, una herramienta es simplemente una función decorada con @tool que recibe un diccionario libre (dict[str, Any]). La ausencia de esquemas de validación formales provoca que los errores de tipo solo se manifiesten en producción cuando el LLM suministra un argumento mal formateado.

3. Pruebas Locales Inviables y Fricción de Desarrollo: Para comprobar si una nueva herramienta funciona, el desarrollador se ve obligado a levantar Docker Compose con PostgreSQL, configurar variables de entorno del LLM, arrancar FastAPI y simular peticiones HTTP completas. Este ciclo lento disuade la escritura de pruebas exhaustivas.

3. Arquitectura del SDK: Contratos Fuertes y Desacoplamiento Absoluto

jg_agente_sdk actúa como la única fuente de verdad (Single Source of Truth) compartida entre los desarrolladores de plugins y el runtime de ejecución.

Diagrama de Desacoplamiento y Topología de Plugins

┌────────────────────────────────────────────────────────────────────────┐
│                        jg_agente_sdk (PyPI: MIT)                       │
│                                                                        │
│   - Contratos Canónicos: AgentTool, ToolResult, BasePlugin, RiskLevel  │
│   - Testing Harness: MockPluginRunner                                  │
│   - CLI Scaffolding: jg                                                │
│   * Cero dependencias de LangGraph, FastAPI, SQLAlchemy ni Docker      │
└───────────────────────────────────┬────────────────────────────────────┘
                                    │
           ┌────────────────────────┴────────────────────────┐
           │ Implementa contratos                            │ Consume contratos
           ▼                                                 ▼
┌──────────────────────────────────────┐  Carga Dinámica  ┌──────────────────────────────────────┐
│        PLUGINS SATÉLITES             │ ───────────────► │       EXECUTION PLANE (CORE)         │
│                                      │  (EntryPoint /   │              jg_agente               │
│ - jg_plugin_conversational           │   PyPI Package)  │                                      │
│ - jg_plugin_filmmaker                │                  │ - CapabilityRegistry                 │
│ - Plugins de DevOps / Infra          │                  │ - SecurityGuard                      │
│                                      │                  │ - GraphBuilder (LangGraph)           │
│ * Cada plugin gestiona sus propias   │                  │ * Core agnóstico: no conoce dominios │
│   dependencias y ciclo de vida       │                  │   ni tiene tools fijas               │
└──────────────────────────────────────┘                  └──────────────────────────────────────┘

Invariantes del Contrato Canónico:

1. Aislamiento de Dependencias: jg_agente_sdk depende exclusivamente de pydantic (v2). No importa si el core usa LangGraph, Celery o un autómata propio: los plugins solo conocen las interfaces del SDK.
2. Validación de Esquema de Entrada y Salida: Cada herramienta declara su esquema de argumentos mediante args_schema basado en BaseModel de Pydantic y devuelve un resultado fuertemente tipado ToolResult.
3. Declaración Explícita de Seguridad y Reversibilidad: Toda herramienta debe catalogar su nivel de riesgo (RiskLevel.LOW, MEDIUM, HIGH, CRITICAL), su impacto colateral (read_only=True/False) y si soporta transacciones compensatorias (reversible=True).

4. Código Fuente Comentado

Especificación del Contrato AgentTool en Pydantic v2

# jg_agente_sdk/jg_sdk/contracts.py

from typing import Callable, Any
from collections.abc import Callable
from pydantic import BaseModel, ConfigDict, Field, model_validator
from enum import StrEnum

class RiskLevel(StrEnum):
    LOW = "low"
    MEDIUM = "medium"
    HIGH = "high"
    CRITICAL = "critical"

class ToolType(StrEnum):
    OBSERVER = "observer"
    ACTUATOR = "actuator"
    VERIFIER = "verifier"
    READ = "read"
    MUTATION = "mutation"

class AgentTool(BaseModel):
    """Contrato formal y descriptor canónico de una herramienta ejecutable."""

    model_config = ConfigDict(arbitrary_types_allowed=True, extra="allow")

    name: str = Field(..., min_length=1, description="Identificador único de la herramienta")
    description: str = Field(..., min_length=1, description="Descripción semántica para el function calling del LLM")
    func: Callable[..., Any] = Field(..., description="Función ejecutable que implementa la herramienta")
    args_schema: type[BaseModel] | None = Field(default=None, description="Esquema Pydantic de validación de argumentos")
    risk_level: RiskLevel = Field(default=RiskLevel.LOW, description="Nivel de riesgo operativo para el SecurityGuard")
    tool_type: ToolType = Field(default=ToolType.ACTUATOR, description="Clasificación funcional de la herramienta")
    read_only: bool = Field(default=False, description="Indica si la herramienta carece de efectos secundarios")
    reversible: bool = Field(default=False, description="Indica si la acción soporta compensación Saga")
    rollback_func: Callable[..., Any] | str | None = Field(
        default=None, description="Función o manejador de compensación inversa obligatoria si reversible=True"
    )

    @model_validator(mode="after")
    def validate_reversible_contract(self) -> "AgentTool":
        """Garantiza que toda herramienta reversible declare su función compensatoria."""
        if self.reversible and not self.rollback_func:
            raise ValueError(
                f"Contrato inválido en '{self.name}': Las herramientas reversibles "
                "deben definir obligatoriamente un 'rollback_func'."
            )
        return self

Estructura Canónica de un Plugin (BasePlugin)

# jg_agente_sdk/jg_sdk/contracts.py

class BasePlugin(ABC):
    """Clase base abstracta que debe implementar todo plugin del ecosistema."""

    name: str = "base_plugin"
    version: str = "1.0.0"

    @abstractmethod
    def register_tools(self) -> list[AgentTool]:
        """Devuelve la lista de herramientas que el plugin aporta al CapabilityRegistry."""
        ...

    def register_engines(self) -> dict[str, Any]:
        """Devuelve motores cognitivos opcionales (planificadores, evaluadores)."""
        return {}

    async def startup(self) -> None:
        """Hook de inicialización asíncrona de recursos (conexiones, pools)."""
        pass

    async def shutdown(self) -> None:
        """Hook de liberación limpia de recursos."""
        pass

5. Scaffolding Rápido con jg

Para acelerar el desarrollo y garantizar el cumplimiento estricto de las convenciones arquitectónicas, el SDK incluye la herramienta de línea de comandos jg.

Creación de un Nuevo Plugin en Segundos:

# Instalación del SDK desde PyPI
pip install jg-agente-sdk

# Generación del scaffolding para un nuevo plugin estándar
jg new mi_plugin_devops --template standard

Este comando genera automáticamente la siguiente estructura de proyecto desacoplada:

mi_plugin_devops/
├── pyproject.toml              # Definición de dependencias aisladas y EntryPoint
├── README.md                   # Documentación técnica del plugin
├── src/
│   └── mi_plugin_devops/
│       ├── __init__.py
│       ├── plugin.py           # Implementación de BasePlugin
│       ├── contracts.py        # Modelos Pydantic específicos de los argumentos
│       └── tools/
│           ├── __init__.py
│           ├── provision.py    # Implementación de AgentTool reversible
│           └── rollback.py     # Lógica de compensación Saga
└── tests/
    └── test_plugin.py          # Pruebas unitarias ultrarrápidas con MockPluginRunner

6. Testing en Milisegundos con MockPluginRunner

Uno de los mayores beneficios de la arquitectura de contratos de jg_agente_sdk es la capacidad de realizar pruebas exhaustivas sin levantar infraestructura externa. MockPluginRunner proporciona un arnés de pruebas (test harness) que simula el ciclo de vida, valida esquemas Pydantic e invoca herramientas directamente.

Suite de Pruebas Unitaria con Pytest (Cero Infraestructura):

# mi_plugin_devops/tests/test_plugin.py

import pytest
from jg_sdk.testing import MockPluginRunner
from jg_sdk.contracts import RiskLevel
from mi_plugin_devops.plugin import DevOpsPlugin

@pytest.mark.asyncio
async def test_devops_plugin_lifecycle_and_tools():
    """Valida el plugin en milisegundos sin levantar bases de datos ni Docker."""
    plugin = DevOpsPlugin()

    # Inicializamos el runner en un contexto asíncrono controlado
    async with MockPluginRunner(plugin) as runner:
        # 1. Verificación de metadatos del plugin
        assert runner.metadata.name == "mi_plugin_devops"
        assert runner.metadata.version == "1.0.0"

        # 2. Verificación de herramientas registradas
        tools = runner.list_tools()
        assert len(tools) >= 1
        
        tool = runner.get_tool("provision_cluster")
        assert tool.risk_level == RiskLevel.HIGH
        assert tool.reversible is True

        # 3. Invocación directa y validación de contrato
        payload = {"cluster_name": "k8s-staging-01", "node_count": 3}
        result = await tool.func(**payload)

        # 4. Aserción sobre el contrato de retorno ToolResult
        assert result.success is True
        assert "k8s-staging-01" in str(result.output)
        assert result.rollback_context is not None
        assert result.rollback_context["cluster_id"] == "cl-staging-01"

Esta prueba se ejecuta en menos de 30 milisegundos, lo que permite integrarla de forma transparente en pipelines de CI/CD locales o pre-commit hooks sin ralentizar el flujo de trabajo del equipo.

7. Consideraciones de Producción, Trade-offs y Límites

1. Límites de Seguridad del SDK: Es indispensable reconocer qué aspectos no cubre el SDK: jg_agente_sdk es una especificación de contratos en memoria; no ejecuta aislamiento a nivel de kernel ni controla permisos de red.

   • Declarar que una herramienta es read_only=True o que opera con RiskLevel.LOW es una declaración de intenciones del desarrollador. La verificación y contención real frente a comportamientos maliciosos o escapes de directorio depende exclusivamente del SecurityGuard del core y del sandboxing del sistema operativo.
   • Del mismo modo, el SDK no valida si la descripción semántica de la herramienta contiene inyecciones de prompts ocultas; la sanitización y evaluación de intenciones corresponde a la máquina de estados de jg_agente.

2. Versionado Semántico y Ruptura de Contratos: Dado que los plugins se cargan dinámicamente mediante EntryPoints o paquetes independientes, una incompatibilidad de versiones entre el core y el SDK puede provocar fallos al arrancar. Por este motivo, el CapabilityRegistry valida estrictamente la versión mayor del SDK en el momento del registro.

8. Reflexión de Arquitectura: La Autonomía de la Periferia

Uno de los errores más comunes al diseñar sistemas extensibles es transferir la complejidad de los casos de uso particulares al núcleo del sistema. La tentación de resolver un problema urgente añadiendo un parámetro especial en el motor central parece inofensiva al principio, pero a medio plazo condena a la plataforma al estancamiento.

La verdadera elegancia arquitectónica radica en construir un núcleo agnóstico, inflexible en sus garantías de seguridad y persistencia, y delegar toda la riqueza funcional en una periferia gobernada por contratos matemáticos. Con jg_agente_sdk, cualquier desarrollador puede concebir, programar y testear una nueva habilidad para el agente en cuestión de minutos, con la certeza de que el core la ejecutará de forma predecible, segura y auditable.

9. Enlaces y Referencias en el Repositorio

• Paquete Canónico del SDK: jg_agente_sdk/jg_sdk/contracts.py
• Harness de Pruebas Unitarias (MockPluginRunner): jg_agente_sdk/jg_sdk/testing.py
• Herramienta CLI de Scaffolding: jg_agente_sdk/jg_sdk/cli.py
• Registro de Capacidades en el Core: jg_agente/app/core/registry.py
• Pruebas de Contratos e Integración: jg_agente_sdk/tests/test_contracts.py