ggr-ai-sdk 0.4.3

SDK Bulletproof para integrar aplicaciones Django Ninja con el GGR AI Gateway vía MCP.

GGR AI SDK

SDK bulletproof para integrar aplicaciones Django Ninja (Spokes) con el GGR AI Gateway vía MCP (Vertex AI Function Calling).

Establece una comunicación bidireccional tolerante a fallos y previene alucinaciones de argumentos del LLM generando automáticamente esquemas JSON a partir de las firmas de tus funciones Python.

Características

• Registro por decorador (@mcp_tool): inspecciona la firma de la función, genera el esquema Pydantic→Vertex y oculta el contexto estático al LLM.
• Inyección de contexto Zero-Trust (MCPContext): las variables sensibles (user_id, tenant_id, …) nunca las decide el LLM; fluyen por el contexto estático.
• Cliente HTTP resiliente: reintentos con backoff exponencial + jitter.
• Manejo de errores estructurado: las herramientas devuelven {"error": ...} en lugar de provocar un 500, habilitando el reintento cognitivo del LLM.
• Runs agénticos durables: cliente start_run / resume_run / approve_run / get_run
  • SSE stream_run, con webhook de resultado terminal y señales ai_run_completed/failed.
• Multimodal y memoria: attachments (Base64) y session_id en el dispatch.
• Idempotencia: deduplica reintentos de Cloud Tasks en /execute (caché de Django).
• Módulo contrib (opcional): seguimiento persistente del ciclo de vida del chat en la BD del Spoke y actualizaciones en tiempo real al frontend vía Server-Sent Events (SSE).

Instalación

pip install ggr-ai-sdk

Uso mínimo

from ggr_ai_sdk import mcp_tool, MCPContext

@mcp_tool(name="cerrar_ticket", description="Cierra un ticket de soporte.")
async def cerrar_ticket(context: MCPContext, ticket_id: int, resolucion: str) -> str:
    user_id = context.user_id  # proviene del contexto estático, no del LLM
    if not user_id:
        return '{"error": "Usuario no identificado en el contexto estático."}'
    # ... lógica de negocio ...
    return f"Ticket {ticket_id} cerrado."

Registra el router del SDK en tu API de Django Ninja:

from ninja import NinjaAPI
from ggr_ai_sdk.routers import mcp_router

api = NinjaAPI()
# Verificados por firma HMAC del Gateway:
#   /execute          ejecución de herramientas MCP
#   /webhook          resultado de un dispatch
#   /webhook/run      resultado terminal de un run durable
api.add_router("/sdk", mcp_router)

Runs agénticos durables

from ggr_ai_sdk import AIGatewayClient, ai_run_completed

client = AIGatewayClient(base_url=settings.AI_GATEWAY_URL, api_key=settings.GGR_AI_SPOKE_API_KEY)

# Inicia un run; el resultado terminal llega por webhook a /sdk/webhook/run.
run = await client.start_run(
    goal="Reagenda todas las citas del cliente 42 a la próxima semana",
    callback_url="https://tu-spoke/api/sdk/webhook/run",
    session_id="conversacion-uuid",  # opcional: memoria entre runs
)

# Reacciona al resultado vía señal de Django.
def on_run_done(sender, payload, **kwargs):
    print(payload.run_id, payload.status, payload.result_text)

ai_run_completed.connect(on_run_done)

El Gateway envía el webhook de run a la callback_url tal cual: debe apuntar a /sdk/webhook/run (distinto del /sdk/webhook del dispatch, cuyo payload es diferente).

Configuración (settings.py del Spoke)

Variable	Descripción
AI_GATEWAY_URL	URL del GGR AI Gateway (local: http://localhost:8001).
AI_CALLBACK_URL	URL del webhook del Spoke al que responde el Gateway.
GGR_AI_SPOKE_API_KEY	API Key compartida con el Gateway (UUID de la ClientApp), usada saliente (X-API-Key al despachar). Obligatoria en producción.
GGR_AI_WEBHOOK_SECRET	Secreto HMAC compartido (ClientApp.webhook_secret del Gateway). El SDK lo usa para verificar la firma (X-Timestamp + X-Signature) de las llamadas entrantes del Gateway a /execute y /webhook. Obligatoria en producción.
GGR_AI_IDEMPOTENCY_TTL_SECONDS	(opcional, def. 86400) TTL del resultado deduplicado por idempotency_key. La deduplicación usa el framework de caché de Django: en producción configura un backend compartido (Redis) para que funcione entre procesos/instancias.

Seguridad entrante (HMAC): el Gateway firma cada llamada al Spoke con hmac_sha256(webhook_secret, f"{timestamp}." + raw_body) y la envía en X-Signature (sha256=…) junto con X-Timestamp. El SDK la verifica en tiempo constante y rechaza firmas inválidas o fuera de la ventana anti-replay (300 s). En DEBUG sin secreto configurado, la verificación se omite con un warning (solo para desarrollo local).

El contrato de payloads (AIRequestPayload / WebhookResponsePayload) y los estados (COMPLETED / FAILED) deben mantenerse en sincronía con el GGR AI Gateway.

Para la integración completa con persistencia y SSE, consulta el manual paso a paso en CLAUDE.md.

Desarrollo

.venv\Scripts\pytest.exe                       # pruebas
.venv\Scripts\ruff.exe check src/ tests/ --fix # lint + formato
.venv\Scripts\mypy.exe src/ tests/             # tipos (modo estricto)