Framework multi-agente abilitato MCP per flussi di lavoro agentici dichiarativi guidati da YAML, utilizzato per l'auditing del codice assistito dall'IA, il triage delle vulnerabilità e la ricerca sulla sicurezza con integrazione CodeQL.
Il Security Lab Taskflow Agent è un framework multi-Agente abilitato MCP per flussi di lavoro agentici dichiarativi basati su YAML.
Costruito sopra l'OpenAI Agents SDK, utilizza Pydantic per la validazione della grammatica e Jinja2 per il rendering dei template.
Il Taskflow Agent sfrutta una grammatica basata su YAML simile a GitHub Workflow per eseguire una serie di attività utilizzando un insieme di Agenti.
La sua proposta di valore principale è quella di essere uno strumento CLI che consente agli utenti di definire e scriptare rapidamente flussi di lavoro agentici senza dover scrivere alcun codice.
Gli Agenti sono definiti tramite personalità, che ricevono un task da completare, dato un insieme di strumenti.
Gli Agenti possono cooperare per completare sequenze di attività attraverso i cosiddetti taskflow.
Puoi trovare una panoramica dettagliata della grammatica dei taskflow qui e taskflow di esempio qui.
┌─────────────────────────────────────────────────────┐
│ 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
L'agente supporta sia l'API Chat Completions che Responses di OpenAI.
Il tipo di API può essere configurato globalmente o per modello in un file 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 per modello può includere:
api_type — "chat_completions" (predefinito) o "responses"endpoint — override dell'URL di base dell'API per questo modellotoken — nome di una variabile d'ambiente contenente la chiave APIIl runner può pilotare tre SDK dietro un'interfaccia comune:
openai_agents (predefinito) — l'OpenAI Agents Python SDK. Supporta
handoff multi-personalità, sia chat_completions che responses
come api_type, temperature, parallel_tool_calls,
exclude_from_context e MCP su stdio, SSE e streamable HTTP.copilot_sdk — il GitHub Copilot Python SDK. Supporta lo streaming,
reasoning_effort, MCP su stdio/SSE/HTTP e il controllo dei permessi
per singolo tool. L'SDK seleziona autonomamente il proprio protocollo di rete per modello, quindi il campo
api_type nel YAML non viene rispettato; gli handoff multi-personalità,
temperature e parallel_tool_calls non sono parimenti disponibili.
I taskflow che utilizzano campi non supportati falliscono al caricamento con un
BackendCapabilityError che indica il campo incriminato.anthropic_sdk — l'Anthropic Python SDK, che pilota la nativa
Messages API (/v1/messages). Supporta lo streaming, il tool calling tramite
MCP e il thinking adattivo con reasoning.effort configurabile
(low, medium, high, max). Gli handoff non sono supportati.
Progettato per l'uso con l'endpoint Anthropic di CAPI; l'autenticazione utilizza
Authorization: Bearer (non x-api-key).Precedenza di selezione (dalla più alta alla più bassa):
backend: per task nel blocco model_settings del task stesso (sovrascrive
il valore a livello di modello per quel singolo task; vedi _resolve_task_model()).backend: per modello nel model_settings della configurazione del modello (consente
backend misti in un singolo taskflow).backend: al livello superiore del documento di configurazione del modello
(predefinito globale).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
Le esecuzioni di Taskflow vengono automaticamente sottoposte a checkpoint a livello di task. Se un task fallisce dopo aver esaurito i tentativi, la sessione viene salvata e può essere ripresa:
** 🤖💾 Session saved: abc123def456
** 🤖💡 Resume with: --resume abc123def456
Riprendi dall'ultimo checkpoint riuscito:
python -m seclab_taskflow_agent --resume abc123def456
Il checkpoint di sessione persiste il valore --model-config fornito dalla CLI (se
presente), quindi le riprese utilizzano la stessa configurazione del modello per impostazione predefinita. Per sovrascrivere
la configurazione del modello alla ripresa, passare --model-config / -m esplicitamente:
python -m seclab_taskflow_agent --resume abc123def456 -m examples.model_configs.responses_api
Le attività non riuscite vengono ritentate automaticamente fino a 3 volte con backoff crescente prima che la sessione venga salvata. I checkpoint di sessione sono memorizzati nella directory dei dati dell'applicazione specifica della piattaforma.
Ogni esecuzione produce un manifest leggibile dalla macchina che riassume ciò che è accaduto:
stato per attività (ok / failed / skipped), i modelli su cui è stata eseguita ogni attività,
la temporizzazione e gli outputs denominati prodotti da ogni attività (inclusi i record
di fan-in per modello per le attività multi-modello). Non contiene endpoint né token.
Il manifest viene scritto in una directory di artefatti con ambito di esecuzione quando un'esecuzione termina o fallisce, e può essere stampato per qualsiasi sessione tramite ID: