Framework multiagente con soporte MCP para flujos de trabajo agénticos declarativos basados en YAML, utilizado para auditoría de código asistida por IA, triaje de vulnerabilidades e investigación de seguridad con integración de CodeQL.
El Agente de Taskflow de Security Lab es un framework multi-Agente habilitado para MCP para flujos de trabajo agénticos declarativos basados en YAML.
Construido sobre el OpenAI Agents SDK, utiliza Pydantic para la validación de gramática y Jinja2 para el renderizado de plantillas.
El Agente de Taskflow aprovecha una gramática basada en YAML similar a GitHub Workflow para realizar una serie de tareas utilizando un conjunto de Agentes.
Su principal propuesta de valor es como una herramienta CLI que permite a los usuarios definir y programar rápidamente flujos de trabajo agénticos sin tener que escribir ningún código.
Los Agentes se definen a través de personalidades, que reciben una tarea a completar, dado un conjunto de herramientas.
Los Agentes pueden cooperar para completar secuencias de tareas a través de los llamados taskflows.
Puedes encontrar una descripción detallada de la gramática de taskflow aquí y ejemplos de taskflows aquí.
┌─────────────────────────────────────────────────────┐
│ CLI (cli.py) │
│ Typer-based entry point: -p, -t, -l, -g, -m, --resume, --lint│
└─────────────────────┬───────────────────────────────┘
│
┌─────────────────────▼───────────────────────────────┐
│ Runner (runner.py) │
│ Taskflow execution loop, model resolution, │
│ template rendering, session checkpointing │
└─────────────────────┬───────────────────────────────┘
│
┌─────────────────────▼───────────────────────────────┐
│ MCP Lifecycle (mcp_lifecycle.py) │
│ Server connection, cleanup, process management │
└─────────────────────┬───────────────────────────────┘
│
┌─────────────────────▼───────────────────────────────┐
│ Agent (agent.py) │
│ TaskAgent wrapper, hooks, OpenAI Agents SDK bridge │
└─────────────────────────────────────────────────────┘
Supporting modules:
models.py — Pydantic v2 grammar models (validation)
session.py — Task-level checkpoint / resume
available_tools.py — YAML resource loader with caching
template_utils.py — Jinja2 template environment
mcp_utils.py — MCP client parameter resolution
mcp_transport.py — MCP transport implementations (stdio, streamable)
mcp_prompt.py — System prompt construction
prompt_parser.py — Legacy prompt argument parser
capi.py — AI API endpoint and token management
path_utils.py — Platform-aware data/log directories
El agente admite tanto la API Chat Completions como la Responses de OpenAI.
El tipo de API se puede configurar de forma global o por modelo en un archivo model_config:
seclab-taskflow-agent:
version: "1.0"
filetype: model_config
api_type: chat_completions # default for all models
models:
gpt_default: gpt-4.1
gpt_responses: gpt-5.1
model_settings:
gpt_responses:
api_type: responses # override for this model
endpoint: https://api.githubcopilot.com
token: CAPI_TOKEN # env var name containing the API key
model_settings por modelo puede incluir:
api_type — "chat_completions" (predeterminado) o "responses"endpoint — anulación de la URL base de la API para este modelotoken — nombre de una variable de entorno que contiene la clave de APIEl runner puede controlar tres SDK tras una interfaz común:
openai_agents (predeterminado) — el SDK de Python de OpenAI Agents. Admite
traspasos multipersonalidad, tanto chat_completions como responses
api_type, temperature, parallel_tool_calls,
exclude_from_context y MCP sobre stdio, SSE y HTTP transmitible.copilot_sdk — el SDK de Python de GitHub Copilot. Admite streaming,
reasoning_effort, MCP sobre stdio/SSE/HTTP y control de permisos
por herramienta. El SDK selecciona su propio protocolo de comunicación por modelo, por lo que el campo
api_type del YAML no se respeta; los traspasos multipersonalidad,
temperature y parallel_tool_calls tampoco están disponibles.
Los taskflows que usan campos no admitidos fallan en tiempo de carga con un
BackendCapabilityError que nombra el campo infractor.anthropic_sdk — el SDK de Python de Anthropic, que controla la API nativa
Messages (/v1/messages). Admite streaming, llamadas a herramientas mediante
MCP y pensamiento adaptativo con reasoning.effort configurable
(low, medium, high, max). No se admiten traspasos.
Diseñado para usarse con el endpoint de Anthropic de CAPI; la autenticación usa
Authorization: Bearer (no x-api-key).Precedencia de selección (de mayor a menor):
backend: por tarea en el propio bloque model_settings de la tarea (anula
el valor a nivel de modelo para esa tarea concreta; véase _resolve_task_model()).backend: por modelo en el model_settings de la configuración del modelo (permite
backends mixtos en un único taskflow).backend: en el nivel superior del documento de configuración del modelo
(predeterminado global).SECLAB_TASKFLOW_BACKEND.openai_agents.seclab-taskflow-agent:
version: "1.0"
filetype: model_config
models:
code_analysis: claude-opus-4.7
general_tasks: gpt-5.4-mini
model_settings:
code_analysis:
api_type: messages
backend: anthropic_sdk
reasoning:
effort: high
general_tasks:
api_type: responses
backend: openai_agents
Las ejecuciones de Taskflow se guardan automáticamente mediante checkpoints a nivel de tarea. Si una tarea falla tras agotar los reintentos, la sesión se guarda y puede reanudarse:
** 🤖💾 Session saved: abc123def456
** 🤖💡 Resume with: --resume abc123def456
Reanudar desde el último checkpoint exitoso:
python -m seclab_taskflow_agent --resume abc123def456
El checkpoint de la sesión conserva el valor de --model-config proporcionado por la CLI (si lo hay), por lo que las reanudaciones usan la misma configuración de modelo por defecto. Para sobrescribir la configuración del modelo al reanudar, pase --model-config / -m explícitamente:
python -m seclab_taskflow_agent --resume abc123def456 -m examples.model_configs.responses_api
Las tareas fallidas se reintentan automáticamente hasta 3 veces con un backoff creciente antes de que se guarde la sesión. Los puntos de control de la sesión se almacenan en el directorio de datos de la aplicación específico de la plataforma.
Cada ejecución produce un manifiesto legible por máquina que resume lo que ocurrió:
estado por tarea (ok / failed / skipped), los modelos con los que se ejecutó cada tarea,
los tiempos y los outputs con nombre que produjo cada tarea (incluidos los registros
de fan-in por modelo para tareas multimodelo). No contiene endpoints ni tokens.
El manifiesto se escribe en un directorio de artefactos con alcance de ejecución cuando una ejecución finaliza o falla, y puede imprimirse para cualquier sesión por ID: