Agente para errores en APIs

¿Alguna vez te has sentido perdido entre miles de líneas de registros intentando encontrar el origen de un fallo crítico?

En la era de la observabilidad, hemos consolidado el monitoreo y los tableros visuales, pero identificar patrones de error sigue siendo una tarea manual, costosa y tediosa para ingenieros y desarrolladores.

En este artículo, exploraremos cómo transformar esta frustración en eficiencia mediante la creación de un agente inteligente basado en LLMs [Large Language Models]. La propuesta de valor es clara: automatizar la extracción y el filtrado de logs para detectar incidentes de forma natural y efectiva, permitiendo que tus equipos se enfoquen en innovar y no solo en resolver incidentes.

A lo largo de este documento, abordaremos los pilares fundamentales para lograrlo: desde la arquitectura de referencia necesaria hasta el paso a paso para implementar una solución que simplifique la causa raíz y potencie el trabajo de arquitectos, desarrolladores y equipos de soporte.

Introducción

En la siguiente sección vamos a explicar cómo desplegar un agente enfocándonos en la búsqueda de patrones de error sobre logs de APIs en **Apigee. Es necesario que cuentes con los siguientes pre requisitos**:

• Un proyecto de GCP
• Una organización de Apigee instalada, puede ser de **evaluación**.
• Conocimientos de git básicos.
• Acceso a tu cuenta de IAM de logging viewer.
• Una **llave** para uso de modelos de LLMs con Google AI Studio.

Arquitectura de referencia

La siguiente arquitectura muestra los componentes que vamos a utilizar para lograr este ejemplo de agente conectándose a Cloud Logging en busca de patrones de error.

Figura 1. Arquitectura de Solución para el Agente

Vamos a utilizar la organización de Apigee para disponer un API con la que podamos generar tráfico de errores buscando errores HTTP 500 y HTTP 401, estos dos patrones se van a registrar en Cloud Logging. Una vez que tengamos estos patrones vamos a buscar que nuestro agente se conecte a los logs y busque sobre un patrón específico de tiempo y sobre un API. El agente no solo debe filtrar los logs, si no que buscamos que nos entregue una explicación de qué está ocurriendo dentro del Proxy.

Instrucciones

En esta sección vas a encontrar los pasos para crear un agente de búsqueda de errores.

Accede a Google Cloud Shell

Figura 2. Acceso a cloud shell

Espera a que se inicie la terminal.

Figura 3. Terminal de cloud shell

Instala ADK, puedes encontrar más información aquí:

pip install google-adk

Paso 1

Clonar el siguiente repositorio.

git clone https://github.com/CarlosGoogleMX/apigee-agents.git

Crear el siguiente archivo.

cd apigee-agents/
nano .env

Crear las siguientes variables

GOOGLE_API_KEY=<your_api_key_from_ai_studio>
GCP_PROJECT=<your_gcp_project>
OAUTH_TOKEN=<your_gcp_ouath_token>

Es importante recordar que el valor de GOOGLE_API_KEY lo vas a obtener de AI studio, ahí podemos generar nuestras llaves, estas llaves nos sirve para llamar el modelo de IA que va a usar nuestro agente. El proyecto de gcp es donde estamos almacenando nuestros logs. Finalmente el token de OAUTH_TOKEN lo podemos obtener con el siguiente comando gcloud auth print-access-token. Este será nuestro token de acceso para consumir las APIs de Google Cloud. El agente va a usar este token para acceder a los proyectos.

Figura 4. Archivo .env

Nota: Validar que nuestro usuario tiene los permisos de Logging Viewer en IAM para que este token sea útil al momento de consumir los logs.

Paso 2

Crea una cuenta de servicio con los permisos de log writer.

gcloud iam service-accounts create sa-example \
- display-name="Example Service Account"

gcloud projects add-iam-policy-binding [PROJECT_ID] \
- member="serviceAccount:sa-example@[PROJECT_ID].iam.gserviceaccount.com" \
- role="roles/logging.logWriter"

Es importante recordar que vamos a necesitar permisos de Service Account User sobre nuestro usuario al momento de desplegar el proxy.

Figura 5. Cuenta de servicio con los permisos de Logs Writter

Importar proxy de ejemplo, tomar el archivo proxy.zip e importarlo en la UI de Apigee. Lo que vamos a encontrar es un proxy de Apigee en el cual agregamos la politíca de logs.

<?xml version="1.0" encoding="UTF-8" standalone="yes"?>
<ProxyEndpoint name="default">
 <Description/>
 <FaultRules/>
 <PreFlow name="PreFlow">
   <Request/>
   <Response/>
 </PreFlow>
 <PostClientFlow name="PostClientFlow">
   <Request/>
   <Response>
     <Step>
       <Name>ML-LOG-INFO</Name>
     </Step>
   </Response>
 </PostClientFlow>
 <Flows/>
 <HTTPProxyConnection>
   <BasePath>/agent-test</BasePath>
   <Properties/>
 </HTTPProxyConnection>
 <RouteRule name="default">
   <TargetEndpoint>default</TargetEndpoint>
 </RouteRule>
</ProxyEndpoint>

Desplegar proxy en algún ambiente, utilizando la cuenta de servicio “sa-example@[PROJECT_ID].iam.gserviceaccount.com” que acabamos de crear.

Figura 6. Despliegue de proxy con la cuenta de servicio

Probar consumo de API

curl https://<your-hostname>/agent-test/1 -v

En este caso por el mock server con el que estamos trabajando esperamos obtener un error HTTP 400. En caso de que envíemos un valor válido como 101, esperamos obtener un error HTTP 502, con esto buscamos encontrar los dos patrones de comportamiento.

Figura 7. Ejemplo de respuesta de CURL

Paso 3

Instalar requerimientos del agente

pip install -r requirements.txt

Figura 8.Instalación de paquetes.

En el archivo agent.py vas a encontrar la lógica de inicialización del agente, de igual manera vas a revisar las instrucciones con las que cuenta el agente y en la línea 4 del archivo vas a encontrar las tools que está utilizando.

import os
from dotenv import load_dotenv
from google.adk.agents.llm_agent import Agent

from .tools import get_cloud_logs, get_current_time

# Load environment variables from .env file
load_dotenv()

root_agent = Agent(
    model='gemini-3.5-flash',
    tools=[get_cloud_logs, get_current_time],
    name='apigee_agents',
    description="An agent that retrieves and analyzes Apigee error logs to diagnose issues.",
    instruction=(
        "You are an expert in analyzing Apigee logs to diagnose issues. "
        "You will need as input the API Name, environment, amount of logs to retrieve"
        "The user may describe time frames in Spanish (e.g., 'hace 3 horas', 'entre la 1 y 2 pm', 'el martes 23'). "
        "You must interpret these time references and convert them into specific start and end times in ISO 8601 format (UTC). "
        "Note that the user's local time is typically Mexico time (UTC-6). You should convert the local time to UTC before passing to the tool. "
        "Before calculating relative time frames, you should call the 'get_current_time' tool to obtain the exact current date and time. Use that value as the base for your calculations. Ensure you use the correct current year (e.g., 2026) and do not assume a past year unless specified. "
        "Then, use the 'get_cloud_logs' tool passing the extracted `start_time` and `end_time` along with other filters to retrieve matching logs containing common errors (4xx and 5xx status codes). "
        "Once you receive the logs, analyze them to identify patterns, such as a spike in a specific error code, "
        "issues from a single client IP, or recurring error messages. "
        "Finally, provide a concise conclusion about the likely cause of the problem based on your analysis."
    ),
)

if __name__ == "__main__":
    print(f"Agent '{root_agent.name}' initialized and ready.")
    print(f"Model configured: {root_agent.model}")
    print(f"Tools available: {[tool.__name__ for tool in root_agent.tools]}")

En el archivo de tools.py puedes leer las funciones con las cuales cuenta el agente para resolver sus tareas. Donde la función principal es get_cloud_logs que nos sirve para extraer los logs de GCP y la función get_current_time que permite saber qué fecha es actualmente y con esta información calcular en la fecha correcta, por ejemplo “hace dos horas”.

import os
import re
import logging
from datetime import datetime, timedelta, timezone
from typing import Optional, Tuple, List, Dict, Any

import google.auth
from google.cloud import logging_v2
from google.oauth2.credentials import Credentials
from google.auth.exceptions import GoogleAuthError

# Configure basic logging
logging.basicConfig(level=logging.INFO, format='%(levelname)s: %(message)s')
logger = logging.getLogger(__name__)

def _clean_log_entry(payload: Any) -> Any:
    """
    Cleans a log entry payload by truncating large fields in the 'logs' array.
    Specifically targets base64 encoded data or large XML/JSON strings.
    """
    if not isinstance(payload, dict):
        return payload

    logs_array = None
    if "logs" in payload:
        logs_array = payload["logs"]
    elif "jsonPayload" in payload and isinstance(payload["jsonPayload"], dict):
        logs_array = payload["jsonPayload"].get("logs")

    if logs_array and isinstance(logs_array, list):
        for log in logs_array:
                    if isinstance(log, dict) and "payload" in log:
                        content = log["payload"]
                        if isinstance(content, str):
                            # First remove massive base64 strings if type property indicates it
                            if '"type":"base64"' in content or '"type": "base64"' in content:
                                content = re.sub(r'("content"\s*:\s*")[^"]{500,}(")', r'\1[BASE64 CONTENT REMOVED]\2', content)
                                log["payload"] = content

                            # Fallback to general truncation if remaining text is still heavy
                            if len(content) > 1000:
                                log["payload"] = content[:1000] + "... [TRUNCATED to prevent token overflow]"
    return payload

def _parse_time_frame(time_frame: str) -> Optional[Tuple[str, str]]:
    return start_time.strftime('%Y-%m-%dT%H:%M:%S.%fZ'), end_time.strftime('%Y-%m-%dT%H:%M:%S.%fZ')

def get_current_time() -> str:
    """
    Returns the current time in ISO 8601 format (UTC).
    """
    return datetime.now(timezone.utc).strftime('%Y-%m-%dT%H:%M:%S.%fZ')

def get_cloud_logs(
    environment: str, 
    start_time: str, 
    end_time: str, 
    api_proxies: Optional[List[str]] = None, 
    api_product: Optional[str] = None, 
    client_id: Optional[str] = None, 
    target_server: Optional[str] = None, 
    n_results: int = 100, 
    error_codes: Optional[List[int]] = None
) -> Dict[str, Any]:
    """
    Retrieves logs with common error status codes from Cloud Logging.
    Allows filtering by environment, time_frame, and optionally api_proxies, api_product, client_id, and target_server.
    """
    try:
        project_id = os.environ.get("GCP_PROJECT")

        # Determine credentials
        token = os.environ.get("OAUTH_TOKEN")
        credentials = None

        try:
            if token:
                logger.info("Using provided OAUTH_TOKEN for authentication.")
                credentials = Credentials(token)
            else:
                logger.info("OAUTH_TOKEN not found. Falling back to Application Default Credentials.")
                credentials, auth_project = google.auth.default()
                project_id = project_id or auth_project
        except GoogleAuthError as e:
            return {"status": "error", "logs": f"Failed to acquire credentials. Please authenticate via `gcloud auth application-default login` or set the OAUTH_TOKEN environment variable. Details: {str(e)}"}

        if not project_id:
            return {"status": "error", "logs": "GCP_PROJECT environment variable not set and could not be inferred from credentials."}

        client = logging_v2.Client(project=project_id, credentials=credentials)

        # Construct the base filter
        filter_parts = [f'jsonPayload.environment="{environment}"']

        if api_proxies:
            proxies_filter = " OR ".join([f'jsonPayload.apiProxy="{proxy}"' for proxy in api_proxies])
            filter_parts.append(f"({proxies_filter})")
        if api_product:
            filter_parts.append(f'jsonPayload.apiProduct="{api_product}"')
        if client_id:
            filter_parts.append(f'jsonPayload.clientId="{client_id}"')
        if target_server:
            filter_parts.append(f'jsonPayload.targetServer="{target_server}"')

        filter_str = " AND ".join(filter_parts)

        # Add time range to filter
        filter_str += f' AND timestamp >= "{start_time}" AND timestamp <= "{end_time}"'

        # Add filter for common error status codes
        if not error_codes:
            error_codes = [500, 502, 503, 504, 403, 401, 400, 404]

        error_filter = " OR ".join([f'jsonPayload.responseCode = {code}' for code in error_codes])
        filter_str += f" AND ({error_filter})"

        logger.info(f"Constructed filter query: {filter_str}")

        # Ensure we strictly limit the returned logs with max_results instead of just controlling the page chunking logic
        entries = client.list_entries(
            filter_=filter_str,
            max_results=n_results,
        )

        logs = [_clean_log_entry(entry.payload) for entry in entries]

        if not logs:
            return {"status": "success", "logs": "No logs found for the given criteria."}

        return {"status": "success", "logs": logs}

    except Exception as e:
        logger.error(f"Error fetching logs: {e}", exc_info=False)
        return {"status": "error", "logs": f"An unexpected error occurred: {e}"}

Para ver los métodos completos.

https://github.com/CarlosGoogleMX/apigee_agents/blob/main/tools.py

Paso 4

Vamos a correr al agente en la terminal con el siguiente comando.

adk web - allow_origins="*"

Si buscamos correr el agente en la terminal lo podemos hacer con el siguiente comando, nos ubicamos en la raíz del directorio de archivos y ejecutamos.

adk run apigee_agents

Una vez tengamos el agente corriendo podemos saludarlo con un “Hola” para que nos explique qué hace.

Figura 9.Agente corriendo en el ADK Web

Paso 5

En este paso vamos a generar los siguientes curls para simular el tráfico requerido, estamos buscando generar dos errores para que nuestro agente logre encontrar el patrón de búsqueda.

Error 400

Con este comando vamos a buscar generar errores 400.

curl https://<your-hostname>/agent-test/1

Error 502

Con este comando vamos a buscar generar errores 502.

curl https://<your-hostname>/agent-test/101

Si regresamos a Cloud Logging podemos ver los logs y el detalle generados.

Figura 10. Ejemplo de los errores en Cloud Logging

Paso 6

• Iniciamos con obtener nuestro token de OAuth siguiendo el comando del paso 1, e inicializamos nuestro agente con el siguiente comando.

adk web - allow_origins="*"

Generamos nuestra consulta

Figura 11. Ejemplo de una consulta

Obtenemos un resultado

Figura 12. Resultado de la consulta

Con esto identificamos los dos patrones de errores que generamos en los logs. En ellos aparecen nuestros errores 502 y 400. Aquí vemos como el agente nos hace un análisis de ambos errores que generamos en cuestión de segundos.

Conclusion

A través de este ejercicio, hemos demostrado cómo un agente inteligente puede transformar la gestión de incidentes, pasando de una búsqueda manual tediosa a una identificación de patrones de error mucho más ágil y eficaz. Aunque este ejemplo se centra en APIs dentro de Apigee, el potencial es mucho mayor.

Este es solo el punto de partida. Imagina escalar esta solución a múltiples equipos de soporte, integrar el uso de Model Context Protocol (MCP) con Cloud Logging o añadir herramientas adicionales para automatizar respuestas, como el envío de alertas a Google Chat y la creación automática de tickets.

La Inteligencia Artificial te puede ayudar a entender qué está fallando, cuál es la causa raíz y darte las herramientas para actuar de inmediato!

Referencias:

• https://adk.dev/get-started/installation/
• https://docs.cloud.google.com/logging/docs/reference/api-overview
• https://aistudio.google.com/api-keys
• https://github.com/CarlosGoogleMX/apigee-agents
• https://docs.cloud.google.com/apigee/docs/api-platform/reference/policies/message-logging-policy