From e3bd48cf4bfb7ce278dd2e6d9461608db5271ac7 Mon Sep 17 00:00:00 2001 From: "cristiano.hoshikawa" Date: Wed, 21 Oct 2026 08:23:55 -0300 Subject: [PATCH] A2A Protocol --- apps/agent_frontend_a2a/Dockerfile | 3 + apps/agent_frontend_a2a/README.md | 133 +++++++++ apps/agent_frontend_a2a/a2a-client.js | 167 ++++++++++++ apps/agent_frontend_a2a/app.js | 225 +++++++++++++++ apps/agent_frontend_a2a/index.html | 52 ++++ apps/agent_frontend_a2a/package.json | 1 + apps/agent_frontend_a2a/styles.css | 1 + apps/agent_frontend_a2a/tests/client.test.js | 86 ++++++ .../tests/integration.test.js | 32 +++ apps/agent_gateway/Dockerfile.a2a | 7 + apps/agent_gateway/README.md | 10 + apps/agent_gateway/app/main.py | 22 +- apps/agent_gateway/app/settings.py | 8 + .../config/backends.a2a.example.yaml | 27 ++ docs/A2A_ASSINCRONO.md | 248 +++++++++++++++++ docs/A2A_IMPLEMENTACAO.md | 145 ++++++++++ examples/a2a_frontend_demo.py | 43 +++ libs/agent_framework/pyproject.toml | 3 + .../src/agent_framework/a2a/__init__.py | 1 + .../src/agent_framework/a2a/client.py | 108 ++++++++ .../src/agent_framework/a2a/progress.py | 10 + .../src/agent_framework/a2a/server.py | 181 ++++++++++++ .../src/agent_framework/config/settings.py | 7 + .../global_supervisor/client.py | 37 ++- .../global_supervisor/models.py | 6 + .../agent_template_backend/.env.a2a.example | 9 + .../agent_template_backend/Dockerfile.a2a | 8 + templates/agent_template_backend/README.md | 7 + templates/agent_template_backend/app/main.py | 39 ++- tests/test_a2a_async_tasks.py | 258 ++++++++++++++++++ tests/test_a2a_http_streaming.py | 81 ++++++ tests/test_a2a_interoperability.py | 188 +++++++++++++ 32 files changed, 2141 insertions(+), 12 deletions(-) create mode 100644 apps/agent_frontend_a2a/Dockerfile create mode 100644 apps/agent_frontend_a2a/README.md create mode 100644 apps/agent_frontend_a2a/a2a-client.js create mode 100644 apps/agent_frontend_a2a/app.js create mode 100644 apps/agent_frontend_a2a/index.html create mode 100644 apps/agent_frontend_a2a/package.json create mode 100644 apps/agent_frontend_a2a/styles.css create mode 100644 apps/agent_frontend_a2a/tests/client.test.js create mode 100644 apps/agent_frontend_a2a/tests/integration.test.js create mode 100644 apps/agent_gateway/Dockerfile.a2a create mode 100644 apps/agent_gateway/config/backends.a2a.example.yaml create mode 100644 docs/A2A_ASSINCRONO.md create mode 100644 docs/A2A_IMPLEMENTACAO.md create mode 100644 examples/a2a_frontend_demo.py create mode 100644 libs/agent_framework/src/agent_framework/a2a/__init__.py create mode 100644 libs/agent_framework/src/agent_framework/a2a/client.py create mode 100644 libs/agent_framework/src/agent_framework/a2a/progress.py create mode 100644 libs/agent_framework/src/agent_framework/a2a/server.py create mode 100644 templates/agent_template_backend/.env.a2a.example create mode 100644 templates/agent_template_backend/Dockerfile.a2a create mode 100644 tests/test_a2a_async_tasks.py create mode 100644 tests/test_a2a_http_streaming.py create mode 100644 tests/test_a2a_interoperability.py diff --git a/apps/agent_frontend_a2a/Dockerfile b/apps/agent_frontend_a2a/Dockerfile new file mode 100644 index 00000000..30c674d8 --- /dev/null +++ b/apps/agent_frontend_a2a/Dockerfile @@ -0,0 +1,3 @@ +FROM nginx:stable-alpine +COPY index.html styles.css app.js a2a-client.js /usr/share/nginx/html/ +EXPOSE 80 diff --git a/apps/agent_frontend_a2a/README.md b/apps/agent_frontend_a2a/README.md new file mode 100644 index 00000000..69544415 --- /dev/null +++ b/apps/agent_frontend_a2a/README.md @@ -0,0 +1,133 @@ +# Frontend A2A para o Agent Framework OCI + +Interface estática, sem build e sem dependências JavaScript de produção. Usa os +endpoints A2A que já existem no pacote: pode falar com o template de backend +ou com o Agent Gateway. O frontend nativo em `apps/agent_frontend` é preservado. + +## Experimentar sem OCI + +Na raiz do framework: + +```sh +python -m pip install -e 'libs/agent_framework[a2a]' uvicorn +python -m uvicorn examples.a2a_frontend_demo:app --host 127.0.0.1 --port 8000 +``` + +Abra **http://localhost:8000** e clique em **Conectar ao serviço**. O exemplo +serve a tela e a API na mesma origem, sem precisar configurar CORS. + +A demonstração usa o **adaptador A2A e o SDK reais**, com a lógica de negócio +simulada. Digite uma mensagem e acompanhe a tarefa. Comandos de demonstração: + +- `/confirmar`: aguarda sua resposta e retoma a mesma Task. +- `/lento`: permite testar o botão de cancelamento. +- `/erro`: demonstra uma Task que termina em falha. + +Se a porta 8000 estiver ocupada, altere a porta ao iniciar e informe o novo +endereço no campo **Endereço do serviço**, ou use `?service=http://localhost:NOVA_PORTA`. + +## Usar com seu agente ou Gateway + +Habilite A2A no serviço, conforme `docs/A2A_ASSINCRONO.md`, e sirva este diretório: + +```sh +python -m http.server 5173 --directory apps/agent_frontend_a2a +``` + +Abra **http://localhost:5173**. Não abra `index.html` como arquivo local, pois a +interface usa módulos JavaScript e requisições HTTP. + +| Uso | Endereço do serviço | Destino na interface | Seleção carregada do Agent Card | +| --- | --- | --- | --- | +| Template de agente | `http://localhost:8000` | Agente | Perfis de `agents.yaml` | +| Agent Gateway | `http://localhost:8010` | Gateway | Backends do catálogo | + +Use **Seleção automática** para deixar a decisão com o serviço. No Gateway, +um backend selecionado é enviado como `message.metadata.backend_id`. Um perfil +opcional dentro dele pode ser informado em **Perfil do agente no backend**. +O adaptador A2A encaminha a seleção ao roteador existente; os contratos nativos +não são alterados. + +O serviço deve permitir a origem da interface em `CORS_ORIGINS`. O valor padrão +do pacote já permite `http://localhost:5173`. Se mudar o host ou a porta, +configure a origem exata, por exemplo: + +```dotenv +CORS_ORIGINS=http://localhost:5173,http://127.0.0.1:5173 +ENABLE_A2A=true +A2A_TASK_MODE=true +``` + +Para publicar com Docker: + +```sh +docker build -t agent-frontend-a2a apps/agent_frontend_a2a +docker run --rm -p 5173:80 agent-frontend-a2a +``` + +O navegador precisa alcançar o endereço informado no formulário. Para uso +externo, informe a URL pública do backend/Gateway, não o nome interno do +container. Uma interface HTTPS deve acessar um serviço HTTPS. + +## O que a interface oferece + +- Descoberta do Agent Card e escolha de versão automática ou explícita (1.0/0.3). +- Envio de mensagem com retorno imediato de Task. +- Acompanhamento por POST SSE com `SubscribeToTask`/`tasks/resubscribe`. +- Recuperação por `GetTask`/`tasks/get` quando o fluxo termina ou cai. +- Consulta periódica como alternativa ao streaming. +- Cancelamento solicitado ao serviço, incluindo tarefas aguardando input. +- Continuação de tarefas em input-required/auth-required. +- Exibição de respostas diretas e dos artefatos de Tasks, com suporte aos + resultados estruturados de múltiplos agentes do framework. +- Contexto de negócio opcional (cliente e contrato/pedido). +- Identificadores da conversa/tarefa preservados em `sessionStorage` para + retomar após recarregar a mesma aba. O conteúdo do chat não é persistido. +- Nova conversa independente, sem reaproveitar uma Task concluída. + +O frontend usa `fetch` para SSE, permitindo enviar os mesmos cabeçalhos de +autenticação nas consultas e no acompanhamento. Em **Contexto e autenticação**, +informe o nome e o valor completo do cabeçalho exigido pelo serviço, por +exemplo `Authorization` / `Bearer ...`, ou `x-api-key` / valor da chave. +Credenciais ficam somente na memória e precisam ser informadas novamente após +recarregar. Não são gravadas em `sessionStorage` nem colocadas na URL. + +Nenhuma operação de envio é repetida automaticamente. Se a resposta do envio +não chegar, a tela informa a incerteza, pois o serviço pode ter aceitado a +mensagem. O botão **Retomar** consulta a mesma Task, quando seu ID é conhecido. +Trocar conexão, tenant, perfil ou backend requer reconectar e inicia outra +conversa quando a identidade da configuração mudou. + +## Limites herdados do serviço + +O frontend não transforma o armazenamento em memória em armazenamento durável. +Uma Task perdida ao reiniciar o serviço não pode ser recuperada pela tela. +O cancelamento local não desfaz efeitos externos já confirmados e não garante +cancelar uma operação já aceita por um agente remoto. + +`auth-required` indica que a tarefa aguarda autorização, mas esta interface não +implementa login OAuth nem aprovação externa automática. Realize o fluxo exigido +pelo agente, atualize a credencial se necessário e retome a tarefa. + +Não há upload de arquivos, push notifications, gRPC ou emissão de tokens do LLM. +A seleção de versão 0.3 é explícita quando o serviço publica um Agent Card 1.0 +mas também aceita o formato 0.3, como ocorre no adaptador do framework. + +## Testes + +Sem bibliotecas JavaScript adicionais, com Node.js 22 ou superior: + +```sh +cd apps/agent_frontend_a2a +node --test tests/client.test.js +``` + +Com a demonstração em execução: + +```sh +A2A_DEMO_URL=http://localhost:8000 node --test tests/*.test.js +``` + +Os testes de integração exercitam o cliente JavaScript contra o SDK real do +framework: A2A 1.0 e 0.3, envio, streaming, consulta, confirmação, cancelamento +e falha. Os testes de interface também foram realizados em navegador local. diff --git a/apps/agent_frontend_a2a/a2a-client.js b/apps/agent_frontend_a2a/a2a-client.js new file mode 100644 index 00000000..6e615f94 --- /dev/null +++ b/apps/agent_frontend_a2a/a2a-client.js @@ -0,0 +1,167 @@ +export const uid = () => globalThis.crypto.randomUUID(); +export const stateName = (state = '') => state.replace(/^TASK_STATE_/, '').toLowerCase().replaceAll('_', '-'); +export const isStopped = state => ['completed', 'failed', 'canceled', 'rejected', 'input-required', 'auth-required'].includes(stateName(state)); +export const needsInput = state => ['input-required', 'auth-required'].includes(stateName(state)); + +export class RpcError extends Error { + constructor(error) { super(error.message || 'O serviço retornou um erro.'); this.name = 'RpcError'; this.code = error.code; } +} +export function serviceURLs(value) { + const url = new URL(value); + if (!['http:', 'https:'].includes(url.protocol) || url.username || url.password || url.search || url.hash) throw new Error('Informe uma URL HTTP ou HTTPS sem credenciais, parâmetros ou fragmentos.'); + const base = url.href.replace(/\/$/, '').replace(/\/a2a$/, ''); + return { base, rpc: `${base}/a2a`, card: `${base}/.well-known/agent-card.json` }; +} +export function normalizeResult(result) { + if (!result || typeof result !== 'object') throw new Error('Resposta A2A inválida.'); + for (const [key, type] of [['task','task'], ['message','message'], ['statusUpdate','status'], ['artifactUpdate','artifact']]) { + if (result[key]) return { type, value: result[key] }; + } + const kinds = { task: 'task', message: 'message', 'status-update': 'status', 'artifact-update': 'artifact' }; + if (kinds[result.kind]) return { type: kinds[result.kind], value: result }; + throw new Error('O serviço não retornou uma mensagem, tarefa ou evento A2A reconhecido.'); +} +export function selectVersion(card, preference = 'auto') { + if (preference !== 'auto') return preference; + const versions = (card.supportedInterfaces || []).filter(i => i.protocolBinding === 'JSONRPC').map(i => i.protocolVersion); + if (versions.some(v => /^1\./.test(v))) return '1.0'; + if (/^0\.3/.test(card.protocolVersion || '') || versions.some(v => /^0\.3/.test(v))) return '0.3'; + throw new Error('O Agent Card não anuncia A2A 1.0 ou 0.3 por JSON-RPC.'); +} + +// SSE over POST/fetch supports authentication headers and both A2A versions. +export async function* parseSSE(body) { + if (!body) throw new Error('O serviço não abriu o fluxo de eventos.'); + const reader = body.getReader(); + const decoder = new TextDecoder(); + let buffer = '', data = [], size = 0; + try { + while (true) { + const { value, done } = await reader.read(); + buffer += done ? decoder.decode() : decoder.decode(value, { stream: true }); + while (true) { + const end = buffer.search(/[\r\n]/); + if (end < 0 || (!done && buffer[end] === '\r' && end === buffer.length - 1)) break; + const line = buffer.slice(0, end); + buffer = buffer.slice(end + (buffer.slice(end, end + 2) === '\r\n' ? 2 : 1)); + if (!line) { + if (data.length) yield data.join('\n'); + data = []; size = 0; + } else if (line.startsWith('data:')) { + const entry = line.slice(5).replace(/^ /, ''); + data.push(entry); size += entry.length; + } + if (size > 4 * 1024 * 1024) throw new Error('Evento recebido excedeu o limite de tamanho.'); + } + if (buffer.length > 4 * 1024 * 1024) throw new Error('Evento recebido excedeu o limite de tamanho.'); + if (done) break; // A partial frame is not an event; the caller can recover with GetTask. + } + } finally { + await reader.cancel().catch(() => {}); + reader.releaseLock(); + } +} + +export class A2AClient { + constructor({ url, version = '1.0', headers = {}, fetchImpl = globalThis.fetch }) { + this.urls = serviceURLs(url); this.version = version; this.fetch = fetchImpl.bind(globalThis); + this.headers = { ...headers, 'A2A-Version': version }; + } + method(v1, v03) { return this.version === '0.3' ? v03 : v1; } + async discover(signal) { + const response = await this.fetch(this.urls.card, { headers: this.headers, credentials: 'omit', signal }); + if (!response.ok) throw new Error(`Não foi possível carregar o Agent Card (HTTP ${response.status}).`); + return response.json(); + } + async rpc(method, params, { signal, timeoutMs = 15000 } = {}) { + const id = uid(); + const timeout = AbortSignal.timeout(timeoutMs); + const response = await this.fetch(this.urls.rpc, { + method: 'POST', headers: { ...this.headers, 'Content-Type': 'application/json' }, credentials: 'omit', + body: JSON.stringify({ jsonrpc: '2.0', id, method, params }), + signal: signal ? AbortSignal.any([signal, timeout]) : timeout, + }); + if (!response.ok) throw new Error(`O serviço retornou HTTP ${response.status}. Confira o endereço e a autenticação.`); + const data = await response.json(); + return this.validate(data, id); + } + validate(data, id) { + if (data.jsonrpc !== '2.0' || data.id !== id) throw new Error('Resposta JSON-RPC inválida.'); + if (data.error) throw new RpcError(data.error); + if (!Object.hasOwn(data, 'result')) throw new Error('Resposta sem resultado A2A.'); + return data.result; + } + async sendMessage(text, { contextId, taskId, tenantId, agentId, backendId, data = {}, signal } = {}) { + const legacy = this.version === '0.3'; + const message = { + messageId: uid(), role: legacy ? 'user' : 'ROLE_USER', contextId, + parts: [{ text }, { data }], metadata: {}, + }; + if (taskId) message.taskId = taskId; + for (const [key, value] of Object.entries({ tenant_id: tenantId, agent_id: agentId, backend_id: backendId })) { + if (value) message.metadata[key] = value; + } + if (legacy) { + message.kind = 'message'; message.parts[0].kind = 'text'; message.parts[1].kind = 'data'; + } + const result = await this.rpc(this.method('SendMessage', 'message/send'), { + message, configuration: legacy ? { blocking: false } : { returnImmediately: true }, + }, { signal }); + return normalizeResult(result); + } + async getTask(id, signal) { return this.rpc(this.method('GetTask', 'tasks/get'), { id }, { signal }); } + async cancelTask(id) { return this.rpc(this.method('CancelTask', 'tasks/cancel'), { id }); } + async *subscribe(id, signal) { + const requestId = uid(); + const response = await this.fetch(this.urls.rpc, { + method: 'POST', credentials: 'omit', signal, + headers: { ...this.headers, 'Content-Type': 'application/json', Accept: 'text/event-stream' }, + body: JSON.stringify({ jsonrpc: '2.0', id: requestId, method: this.method('SubscribeToTask', 'tasks/resubscribe'), params: { id } }), + }); + if (!response.ok) throw new Error(`Não foi possível acompanhar os eventos (HTTP ${response.status}).`); + if (!(response.headers.get('content-type') || '').includes('text/event-stream')) { + this.validate(await response.json(), requestId); + throw new Error('O serviço não retornou um fluxo SSE.'); + } + for await (const frame of parseSSE(response.body)) { + yield normalizeResult(this.validate(JSON.parse(frame), requestId)); + } + } +} + +export class TaskView { + constructor(task = {}) { this.task = structuredClone(task); this.task.artifacts ||= []; } + apply(event) { + const value = event.value; + if (event.type === 'task') { + if (this.task.id && value.id !== this.task.id) throw new Error('Evento de outra tarefa.'); + this.task = structuredClone(value); this.task.artifacts ||= []; + } + else if (event.type === 'status') { + if (this.task.id && value.taskId !== this.task.id) throw new Error('Evento de outra tarefa.'); + this.task.id = value.taskId; this.task.contextId = value.contextId; this.task.status = value.status; + } else if (event.type === 'artifact') { + if (this.task.id && value.taskId !== this.task.id) throw new Error('Evento de outra tarefa.'); + const index = this.task.artifacts.findIndex(a => a.artifactId === value.artifact.artifactId); + if (index < 0) this.task.artifacts.push(structuredClone(value.artifact)); + else if (value.append) this.task.artifacts[index].parts.push(...structuredClone(value.artifact.parts || [])); + else this.task.artifacts[index] = structuredClone(value.artifact); + } + return this.task; + } +} + +export function displayResponse(value, depth = 0) { + if (!value || depth > 5) return null; + const task = value.task || value; + const parts = task.artifacts?.at(-1)?.parts || task.status?.message?.parts || value.message?.parts || value.parts || []; + const native = parts.find(p => p.data && typeof p.data === 'object' && ('text' in p.data || 'speak' in p.data || 'metadata' in p.data))?.data; + if (native) { + const remote = native.metadata?.a2a_result; + return (remote && displayResponse(remote, depth + 1)) || native; + } + const text = parts.filter(p => typeof p.text === 'string').map(p => p.text).join(''); + if (text) return { text }; + const structured = parts.filter(p => 'data' in p).map(p => p.data); + return structured.length ? { text: JSON.stringify(structured.length === 1 ? structured[0] : structured, null, 2) } : null; +} diff --git a/apps/agent_frontend_a2a/app.js b/apps/agent_frontend_a2a/app.js new file mode 100644 index 00000000..705d6a70 --- /dev/null +++ b/apps/agent_frontend_a2a/app.js @@ -0,0 +1,225 @@ +import { A2AClient, TaskView, displayResponse, isStopped, needsInput, selectVersion, stateName, uid } from './a2a-client.js'; + +const $ = id => document.getElementById(id); +const value = id => $(id).value.trim(); +const labels = { submitted:'Tarefa recebida', working:'Agente trabalhando', completed:'Concluída', failed:'Falha no processamento', canceled:'Tarefa cancelada', rejected:'Solicitação recusada', 'input-required':'Aguardando sua resposta', 'auth-required':'Aguardando autorização' }; +const eventLabels = { 'flow.start':'Fluxo iniciado', 'session.upserted':'Sessão preparada', 'message.received':'Mensagem recebida', 'workflow.started':'Workflow iniciado', 'workflow.completed':'Workflow concluído', 'message.responded':'Resposta disponível', 'flow.end':'Fluxo finalizado', 'a2a.backend.status':'Status do backend atualizado' }; +const state = { client:null, card:null, contextId:uid(), view:new TaskView(), connected:false, busy:false, epoch:0, controller:null, turn:null, events:0, lastState:'', scope:null }; +const savedKey = 'framework-a2a-session-v1'; +const connectionFields = ['serviceUrl','target','version','skill','tenant','tracking','customerKey','contractKey','agentOverride','authHeader','authValue']; +const messageGroups = new Map(); + +function notice(text = '') { $('notice').textContent = text; $('notice').hidden = !text; } +function setStatus(text, kind = '') { $('taskStatus').textContent = text; $('taskDot').className = `dot ${kind}`; } +function setBusy(busy) { state.busy = busy; document.body.setAttribute('aria-busy', String(busy)); controls(); } +function controls() { + const active = state.view.task.id && !isStopped(state.view.task.status?.state); + const allowSend = state.connected && !state.busy && !active; + $('message').disabled = !allowSend; $('sendButton').disabled = !allowSend; + $('cancelButton').disabled = !state.connected || !(active || needsInput(state.view.task.status?.state)); + $('resumeButton').disabled = !state.connected || state.busy || !state.view.task.id; + $('newConversation').disabled = state.busy; + $('connectButton').disabled = state.busy; + for (const id of connectionFields) $(id).disabled = state.busy; +} +function log(text) { + const item = document.createElement('li'); + const time = document.createElement('time'); time.textContent = new Date().toLocaleTimeString('pt-BR'); + item.append(time, document.createTextNode(text)); $('activity').append(item); + while ($('activity').children.length > 80) $('activity').firstChild.remove(); + $('eventCount').textContent = `${++state.events} eventos`; +} +function saveSession() { + // Only identifiers and connection settings: no credentials or message content. + try { sessionStorage.setItem(savedKey, JSON.stringify({ + serviceUrl: state.client?.urls.base, version:state.client?.version, target:value('target'), + contextId:state.contextId, taskId:state.view.task.id, taskState:state.view.task.status?.state, + tenant:value('tenant'), skill:value('skill'), agentOverride:value('agentOverride'), + })); } catch { /* Storage can be unavailable in private browsing. */ } +} +function ids() { + $('taskId').textContent = state.view.task.id || '—'; $('taskShort').textContent = state.view.task.id ? `#${state.view.task.id.slice(0,8)}` : ''; + $('contextId').textContent = state.contextId; $('contextShort').textContent = `Conversa ${state.contextId.slice(0,8)}`; +} +function stopWatcher() { ++state.epoch; state.controller?.abort(); state.controller = null; } +function newConversation() { + stopWatcher(); state.contextId = uid(); state.view = new TaskView(); state.turn = null; state.lastState = ''; messageGroups.clear(); + $('chat').replaceChildren(); $('activity').replaceChildren(); state.events = 0; $('eventCount').textContent='0 eventos'; notice(); + setBusy(false); ids(); setStatus(state.connected ? 'Pronto para conversar' : 'Pronto para conectar'); saveSession(); $('message').focus(); +} +function userMessage(text) { + const div = document.createElement('div'); div.className='message user'; div.textContent=text; + $('welcome')?.remove(); $('chat').append(div); scrollChat(); +} +function agentMessages(response) { + const metadata = response.metadata || {}; + const candidates = response.supervisor_results || metadata.supervisor_results || response.agent_responses || metadata.agent_responses; + if (Array.isArray(candidates)) { + const parts = candidates.map(p => ({ agent:p.agent || p.agent_id || p.name || 'Agente', text:p.answer ?? p.text ?? p.message ?? p.response ?? p.content, primary:p.primary })); + const valid = parts.filter(p => p.text != null && String(p.text).trim()); + if (valid.length) return valid.sort((a,b)=>Number(Boolean(a.primary))-Number(Boolean(b.primary))); + } + const text = response.text ?? response.speak ?? response.content ?? ''; + return [{ agent:metadata.active_agent || metadata.agent_id || state.card?.name || 'Agente', text }]; +} +function scrollChat() { $('chat').scrollTop = $('chat').scrollHeight; } +function renderResponse(response) { + if (!response) return; + $('welcome')?.remove(); state.turn ||= uid(); + let group = messageGroups.get(state.turn); + if (!group) { group=document.createElement('section'); group.className='message agent'; group.setAttribute('aria-label','Resposta do agente'); messageGroups.set(state.turn,group); $('chat').append(group); } + const signature=JSON.stringify(response); if (group.dataset.signature === signature) return; + group.dataset.signature=signature; group.replaceChildren(); + for (const part of agentMessages(response)) { + const section=document.createElement('section'); section.className='agent-section'; + const header=document.createElement('div'); header.className='agent-header'; + const avatar=document.createElement('span'); avatar.className='agent-avatar'; avatar.textContent='✳'; avatar.setAttribute('aria-hidden','true'); + const name=document.createElement('span'); name.textContent=String(part.agent).replaceAll('_',' '); + const body=document.createElement('div'); body.className='agent-body'; body.textContent=typeof part.text==='string' ? part.text : JSON.stringify(part.text,null,2); + header.append(avatar,name); section.append(header,body); group.append(section); + } + scrollChat(); +} +function updateTask(event) { + const task=state.view.apply(event); + if (task.contextId) state.contextId=task.contextId; + const current=stateName(task.status?.state); + if (current && current!==state.lastState) { log(labels[current] || current); state.lastState=current; } + if (event.type==='status' && event.value.metadata?.framework_event) log(eventLabels[event.value.metadata.framework_event] || event.value.metadata.framework_event); + setStatus(labels[current] || 'Acompanhando tarefa', ['failed','rejected'].includes(current) ? 'error' : isStopped(current) ? '' : 'busy'); + if (event.type==='artifact' || isStopped(current)) renderResponse(displayResponse(task)); + ids(); saveSession(); controls(); +} +function sleep(ms,signal) { + return new Promise((resolve,reject)=>{ + if (signal.aborted) { reject(new DOMException('Interrompido','AbortError')); return; } + const abort=()=>{clearTimeout(timer);reject(new DOMException('Interrompido','AbortError'));}; + const timer=setTimeout(()=>{signal.removeEventListener('abort',abort);resolve();},ms); + signal.addEventListener('abort',abort,{once:true}); + }); +} +async function watchTask() { + stopWatcher(); const epoch=state.epoch; const controller=new AbortController(); state.controller=controller; + const signal=controller.signal; const id=state.view.task.id; setBusy(true); + try { + if (value('tracking')==='stream' && state.card?.capabilities?.streaming && !isStopped(state.view.task.status?.state)) { + try { + log('Conectando ao fluxo da tarefa'); + for await (const event of state.client.subscribe(id,signal)) { + if (epoch!==state.epoch) return; updateTask(event); + if (isStopped(state.view.task.status?.state)) break; + } + } catch (error) { + if (signal.aborted) return; + log('Fluxo interrompido; recuperando por consulta'); + } + } + // Always obtain the authoritative final snapshot; also recovers a fast Task + // that completed before SubscribeToTask or a dropped/truncated SSE stream. + let failures=0; + while (true) { + try { + const task=await state.client.getTask(id,signal); + if (epoch!==state.epoch) return; + updateTask({type:'task',value:task}); failures=0; + if (isStopped(task.status?.state)) break; + } catch (error) { + if (signal.aborted) return; + if (++failures>=3 || error.code) throw error; + log('Conexão indisponível; tentando consultar novamente'); + } + await sleep(failures ? failures*1000 : 900,signal); + } + } catch (error) { + if (!signal.aborted && epoch===state.epoch) { + notice(`Acompanhamento pausado: ${error.message} Use Retomar para consultar a mesma tarefa. Nenhuma mensagem foi reenviada.`); + setStatus('Acompanhamento pausado','error'); log('Acompanhamento pausado'); + } + } finally { + if (epoch===state.epoch) { setBusy(false); $('message').focus(); } + } +} +function authHeaders() { + const secret=$('authValue').value; + if (!secret) return {}; + const name=value('authHeader'); + if (!/^[!#$%&'*+.^_`|~0-9A-Za-z-]+$/.test(name) || /[\r\n]/.test(secret)) throw new Error('Cabeçalho de autenticação inválido.'); + if (['content-type','accept','a2a-version','host','origin','cookie'].includes(name.toLowerCase())) throw new Error('Use Authorization ou o cabeçalho de autenticação configurado no serviço.'); + return {[name]:secret}; +} +$('connectionForm').addEventListener('submit',async event=>{ + event.preventDefault(); notice(); setBusy(true); setStatus('Conectando…','busy'); + try { + const candidate=new A2AClient({url:value('serviceUrl'),version:value('version')==='0.3'?'0.3':'1.0',headers:authHeaders()}); + const card=await candidate.discover(AbortSignal.timeout(12000)); candidate.version=selectVersion(card,value('version')); candidate.headers['A2A-Version']=candidate.version; + const scope=JSON.stringify([candidate.urls.base,candidate.version,value('target'),value('tenant'),value('skill'),value('agentOverride')]); + if (state.scope && scope!==state.scope) newConversation(); + state.scope=scope; state.client=candidate; state.card=card; state.connected=true; + const previous=value('skill'); $('skill').replaceChildren(new Option('Seleção automática','')); + for (const skill of card.skills || []) $('skill').append(new Option(skill.name || skill.id,skill.id)); + $('skill').value=[...$('skill').options].some(o=>o.value===previous)?previous:''; + $('serviceName').textContent=card.name || 'Serviço A2A'; $('connectionDetail').textContent=`${(card.skills || []).length === 1 ? '1 opção disponível' : `${(card.skills || []).length} opções disponíveis`} · ${card.capabilities?.streaming ? 'Atualizações em tempo real' : 'Consulta de tarefas'}`; + $('connectionDot').className='dot'; $('versionBadge').textContent=`A2A ${candidate.version}`; + setStatus(state.view.task.id?'Conectado · tarefa disponível para consulta':'Pronto para conversar'); + log('Conectado ao serviço'); ids(); saveSession(); + } catch(error) { + state.connected=false; $('connectionDot').className='dot error'; $('connectionDetail').textContent='Não foi possível conectar.'; + setStatus('Falha na conexão','error'); notice(`${error.message} Se o navegador bloquear a conexão, confira CORS_ORIGINS no serviço e use HTTP/HTTPS para servir esta página.`); + } finally { setBusy(false); } +}); +$('messageForm').addEventListener('submit',async event=>{ + event.preventDefault(); const text=value('message'); if(!text || !state.connected || $('sendButton').disabled) return; + notice(); const previousTask=state.view.task; const resumeId=needsInput(previousTask.status?.state)?previousTask.id:undefined; + state.turn=uid(); userMessage(text); $('message').value=''; $('message').style.height='auto'; setBusy(true); setStatus('Enviando mensagem…','busy'); + const messageId=uid(); + const business_context={customer_key:value('customerKey')||null,contract_key:value('contractKey')||null,interaction_key:messageId,session_key:state.contextId}; + const data={message_id:messageId,business_context,context:{business_context}}; + if(value('customerKey')) data.customer_key=value('customerKey'); + if(value('contractKey')) data.contract_key=value('contractKey'); + try { + const gateway=value('target')==='gateway'; + const result=await state.client.sendMessage(text,{contextId:state.contextId,taskId:resumeId,tenantId:value('tenant')||'default', + agentId:gateway?value('agentOverride'):value('skill'),backendId:gateway?value('skill'):undefined,data}); + if(result.type==='message') { + state.view=new TaskView(); if(result.value.contextId) state.contextId=result.value.contextId; + renderResponse(displayResponse(result.value)); setStatus('Resposta recebida'); ids(); saveSession(); setBusy(false); + } else if(result.type==='task') { + state.view=new TaskView(); state.lastState=''; updateTask(result); log(`Tarefa ${state.view.task.id.slice(0,8)} recebida`); await watchTask(); + } else throw new Error('Resposta de envio inesperada.'); + } catch(error) { + notice(`Não foi possível confirmar o envio: ${error.message} Ele pode ter sido aceito pelo serviço; não houve reenvio automático.`); + setStatus('Envio não confirmado','error'); setBusy(false); + } +}); +$('resumeButton').addEventListener('click',()=>{notice();state.turn ||= uid();watchTask();}); +$('cancelButton').addEventListener('click',async()=>{ + const id=state.view.task.id; if(!id || !state.connected) return; + stopWatcher(); setBusy(true); setStatus('Solicitando cancelamento…','busy'); + try { updateTask({type:'task',value:await state.client.cancelTask(id)});log('Cancelamento confirmado pelo serviço');notice(); } + catch(error) { + try {updateTask({type:'task',value:await state.client.getTask(id)});} catch { /* Keep the last known Task. */ } + notice(`O serviço não confirmou o cancelamento: ${error.message} Consulte a tarefa com Retomar.`); + } finally {setBusy(false);} +}); +$('newConversation').addEventListener('click',newConversation); +$('message').addEventListener('keydown',event=>{if(event.key==='Enter'&&!event.shiftKey&&!event.isComposing){event.preventDefault();$('messageForm').requestSubmit();}}); +$('message').addEventListener('input',()=>{ $('message').style.height='auto';$('message').style.height=`${Math.min($('message').scrollHeight,150)}px`; }); +for(const button of document.querySelectorAll('[data-prompt]')) button.addEventListener('click',()=>{if(!$('message').disabled){$('message').value=button.dataset.prompt;$('message').focus();}else notice('Conecte ao serviço para iniciar a conversa.');}); +$('target').addEventListener('change',()=>{ $('skillLabel').firstChild.textContent=value('target')==='gateway'?'Backend no Gateway':'Agente'; }); +for(const id of ['serviceUrl','version','target','tenant','skill','agentOverride','authHeader','authValue']) $(id).addEventListener('change',()=>{ + if(state.connected){state.connected=false;$('connectionDot').className='dot muted';setStatus('Configuração alterada · conecte novamente');controls();} +}); +try { + const saved=JSON.parse(sessionStorage.getItem(savedKey)||'null'); + if(saved?.serviceUrl){ + $('serviceUrl').value=saved.serviceUrl;$('version').value=saved.version||'auto';$('target').value=saved.target||'backend';$('tenant').value=saved.tenant||'default';$('agentOverride').value=saved.agentOverride||''; + if(saved.skill)$('skill').append(new Option(saved.skill,saved.skill));$('skill').value=saved.skill||''; + state.contextId=saved.contextId||uid();state.view=new TaskView(saved.taskId?{id:saved.taskId,contextId:state.contextId,status:{state:saved.taskState}}:{}); + state.scope=JSON.stringify([saved.serviceUrl,saved.version,saved.target||'backend',saved.tenant||'default',saved.skill||'',saved.agentOverride||'']); + if(saved.taskId)notice('Uma tarefa desta aba está disponível. Conecte ao mesmo serviço e clique em Retomar para consultar o resultado.'); + } +} catch { /* An unavailable or invalid saved session starts a fresh conversation. */ } +const requestedService=new URLSearchParams(location.search).get('service'); +if(requestedService)$('serviceUrl').value=requestedService; +else if(!state.scope && location.port==='8000')$('serviceUrl').value=location.origin; +$('target').dispatchEvent(new Event('change'));ids();controls(); diff --git a/apps/agent_frontend_a2a/index.html b/apps/agent_frontend_a2a/index.html new file mode 100644 index 00000000..7746e584 --- /dev/null +++ b/apps/agent_frontend_a2a/index.html @@ -0,0 +1,52 @@ + + + + + Agent Workspace · A2A + + +
+ +
+
SEU ESPAÇO DE TRABALHO

Converse com seus agentes

+
+
Pronto para conectar
+ +
+
+ +
+
CONVERSAS QUE VIRAM AÇÕES

O próximo passo
começa com uma mensagem.

Conecte um agente ou o Gateway do framework.
Acompanhe cada tarefa até a resposta.

+
+
Enter envia · Shift + Enter quebra a linhaConversa nova
+
Atividade da tarefa 0 eventos
    Contexto
    —
    Tarefa
    —
    +
    +
    + + + diff --git a/apps/agent_frontend_a2a/package.json b/apps/agent_frontend_a2a/package.json new file mode 100644 index 00000000..30a03139 --- /dev/null +++ b/apps/agent_frontend_a2a/package.json @@ -0,0 +1 @@ +{"name":"agent-framework-a2a-frontend","private":true,"type":"module","scripts":{"test":"node --test tests/*.test.js"}} diff --git a/apps/agent_frontend_a2a/styles.css b/apps/agent_frontend_a2a/styles.css new file mode 100644 index 00000000..1f65c3c9 --- /dev/null +++ b/apps/agent_frontend_a2a/styles.css @@ -0,0 +1 @@ +:root{font-family:Inter,ui-sans-serif,system-ui,-apple-system,BlinkMacSystemFont,"Segoe UI",sans-serif;color:#24342d;background:#f8faf7;font-synthesis:none;--green:#285b44;--line:#e0e7df;--muted:#6c7c70}*{box-sizing:border-box}body{margin:0}button,input,select,textarea{font:inherit}button{cursor:pointer;transition:background .15s}button:disabled{cursor:default;opacity:.45}button:focus-visible,input:focus-visible,select:focus-visible,textarea:focus-visible,summary:focus-visible{outline:3px solid #a4c4af;outline-offset:3px}button{border:0}.workspace{display:grid;grid-template-columns:300px minmax(0,1fr);min-height:100vh}.sidebar{padding:30px 24px 18px;background:#f0f4ed;border-right:1px solid var(--line);display:flex;flex-direction:column;gap:17px}.brand{display:flex;gap:11px;align-items:center;text-decoration:none;color:inherit;font-size:17px;font-weight:650;margin:1px 0 26px}.brand-mark{width:37px;height:39px;border:1px solid #365c45;border-radius:12px 12px 12px 3px;display:grid;place-items:center;background:var(--green);color:#fff;font-family:Georgia,serif;font-size:31px}.brand small{display:block;font-size:9px;letter-spacing:1.6px;margin-top:5px;color:var(--muted)}.sidebar-heading{display:flex;justify-content:space-between;align-items:center;font-size:10px;letter-spacing:1.5px;font-weight:650;color:#6e7c6d}.badge{border:1px solid #ccd9ca;padding:3px 6px;border-radius:4px;font-size:9px;letter-spacing:.3px}label{display:flex;flex-direction:column;gap:7px;font-size:11px;font-weight:550;line-height:1.4}.connection-form{display:grid;gap:15px}input,select{width:100%;min-width:0;border:1px solid #d5dfd1;background:#fffefb;border-radius:7px;padding:10px;color:#34473c;font-size:12px}input:disabled,select:disabled{opacity:.55}.two-columns{display:grid;grid-template-columns:1.15fr 1fr;gap:10px}.primary{display:flex;justify-content:space-between;align-items:center;background:var(--green);color:#fff;border-radius:7px;padding:12px 13px;font-size:12px}.primary:hover{background:#1e4935}.service-card{border:1px solid #dce5d7;border-radius:7px;padding:12px;display:flex;align-items:flex-start;gap:9px;background:#f8faf5}.service-card strong{font-size:11px;font-weight:600}.service-card p{font-size:10px;color:var(--muted);line-height:1.65;margin:5px 0 0}.dot{height:7px;width:7px;flex-shrink:0;display:inline-block;background:#548366;border-radius:100%;margin-top:5px}.dot.muted{background:#b2bfb0}.dot.busy{background:#c19540;box-shadow:0 0 0 3px #c1954015}.dot.error{background:#be6553}.advanced{border-top:1px solid var(--line);padding-top:15px}.advanced summary{font-size:11px;cursor:pointer;margin-bottom:13px}.advanced label{margin-bottom:12px}.hint{font-size:10px;color:var(--muted);line-height:1.7}.sidebar-footer{margin-top:auto;padding-top:26px;font-size:9px;color:#80907d;display:flex;align-items:center;gap:7px}.sidebar-footer .dot{height:5px;width:5px;margin:0}.main{min-width:0;display:flex;flex-direction:column;padding:36px clamp(25px,5vw,85px) 20px;height:100vh;overflow:auto}.topbar{display:flex;justify-content:space-between;gap:15px;align-items:center;padding-bottom:26px}.eyebrow{font-size:9px;letter-spacing:1.9px;color:#81917e;font-weight:600}h1{font-family:Georgia,"Times New Roman",serif;font-weight:400;font-size:27px;letter-spacing:-.5px;margin:8px 0 0}.secondary{border:1px solid var(--line);border-radius:6px;background:transparent;padding:10px 12px;font-size:11px;white-space:nowrap;color:#4f6355}.secondary:hover{background:#edf2e9}.task-strip{min-height:49px;display:flex;align-items:center;gap:12px;border-block:1px solid var(--line);padding:7px 0}.status-pill{display:flex;align-items:center;gap:8px;font-size:11px}.status-pill .dot{margin:0}.task-id{font-family:ui-monospace,monospace;font-size:10px;color:#99a593}.task-actions{margin-left:auto;display:flex;gap:14px}.text-button{background:none;color:var(--green);padding:6px 0;font-size:10px}.danger{color:#9b5c4c}.notice{font-size:12px;background:#fff5e9;color:#7e5635;border:1px solid #ead7bb;border-radius:7px;padding:12px 15px;margin-top:14px;line-height:1.6}.chat{flex:1;min-height:280px;overflow-y:auto;padding:30px 0 20px;scroll-behavior:smooth}.welcome{max-width:480px;margin:clamp(15px,6vh,65px) auto 40px;display:flex;flex-direction:column;align-items:center;text-align:center}.welcome-icon{display:grid;place-items:center;color:#557953;background:#edf3e7;border:1px solid #dce8d5;border-radius:16px;width:54px;height:54px;font-size:35px;margin-bottom:26px}.welcome h2{font-family:Georgia,"Times New Roman",serif;font-size:clamp(28px,3.2vw,42px);font-weight:400;line-height:1.14;letter-spacing:-1.1px;margin:17px 0}.welcome p{color:#7b897b;font-size:12px;line-height:1.9;margin:1px 0 28px}.suggestions{display:flex;gap:10px}.suggestions button{border:1px solid var(--line);padding:11px 13px;font-size:10px;background:#fffefb;border-radius:7px;color:#5c705e}.suggestions button span{margin-left:17px;color:#94a38e}.composer-area{padding-top:10px}.composer{display:flex;align-items:flex-end;gap:15px;background:#fffefb;border:1px solid #d7e0d1;box-shadow:0 3px 18px #34502305;border-radius:12px;padding:12px 13px 12px 17px}.composer textarea{width:100%;resize:none;background:none;border:0;outline:none;font-size:12px;line-height:1.7;min-height:32px;max-height:150px;padding:5px 0;color:#304a37}.composer textarea::placeholder{color:#a0ac9c}.send-button{border-radius:8px;background:var(--green);color:white;min-width:34px;height:34px;font-size:21px}.composer-footer{display:flex;justify-content:space-between;gap:15px;font-size:9px;color:#99a493;padding:11px 2px 15px}.activity-panel{border-top:1px solid var(--line);padding-top:13px;font-size:10px;color:var(--muted)}.activity-panel summary{cursor:pointer;display:flex;justify-content:space-between;gap:20px}.activity-panel summary::before{content:'↳';margin-right:5px}.activity-panel summary span{margin-left:auto;color:#97a38e}.activity-panel ol{padding:5px 0 0 20px;max-height:140px;overflow:auto}.activity-panel li{padding:5px 0}.activity-panel time{color:#9aa492;margin-right:10px;font-variant-numeric:tabular-nums}.activity-panel dl{font-family:ui-monospace,monospace;font-size:9px;display:grid;grid-template-columns:65px minmax(0,1fr);gap:5px}.activity-panel dd{margin:0;overflow-wrap:anywhere}.message{margin:0 0 25px;max-width:86%;font-size:13px;line-height:1.85}.message.user{margin-left:auto;background:#e8efe3;border:1px solid #dce6d5;padding:12px 17px;border-radius:12px 12px 3px 12px;max-width:75%;white-space:pre-wrap;overflow-wrap:anywhere}.agent-header{display:flex;align-items:center;gap:9px;margin-bottom:10px;font-size:11px;font-weight:650;color:#526c52}.agent-avatar{display:grid;place-items:center;border:1px solid #d3dfce;background:#eff4e9;border-radius:7px;width:26px;height:26px;font-size:15px}.agent-body{white-space:pre-wrap;overflow-wrap:anywhere;color:#354c3c;padding-left:35px}.agent-section+.agent-section{margin-top:18px}.agent-body pre{white-space:pre-wrap}body[aria-busy="true"] .task-strip .dot.busy{animation:pulse 1.5s infinite}@keyframes pulse{50%{opacity:.45}}.sr-only{position:absolute;width:1px;height:1px;padding:0;overflow:hidden;clip:rect(0,0,0,0);white-space:nowrap;border:0}[hidden]{display:none!important}@media(min-width:1500px){.main{padding-inline:max(65px,calc((100vw - 1200px)/2))}}@media(max-width:900px){.workspace{grid-template-columns:255px minmax(0,1fr)}.sidebar{padding:24px 18px}.main{padding:27px 25px 18px}.topbar{align-items:flex-start}.topbar h1{font-size:25px}.task-id{display:none}.suggestions{flex-direction:column}.message{max-width:100%}}@media(max-width:650px){.workspace{display:block}.sidebar{padding:18px 20px;gap:13px;border-right:0;border-bottom:1px solid var(--line)}.brand{margin:0 0 8px}.connection-form{grid-template-columns:1fr 1fr;align-items:end}.connection-form>label{grid-column:1/-1}.connection-form .two-columns{grid-template-columns:1fr 1fr}.sidebar>label{display:flex}.advanced{margin-top:0}.sidebar-footer{display:none}.main{height:auto;min-height:75vh;padding:24px 20px}.chat{max-height:65vh;min-height:310px}.topbar{padding-bottom:20px}.topbar h1{font-size:24px}.topbar .secondary{font-size:0}.topbar .secondary::after{content:'＋ Nova';font-size:11px}.welcome{margin-top:25px}.welcome h2{font-size:32px}.suggestions{flex-direction:row}.suggestions button{font-size:9px;padding:10px}.task-actions{gap:9px}.status-pill{font-size:10px}.composer-footer{font-size:8px}.message.user{max-width:90%}} diff --git a/apps/agent_frontend_a2a/tests/client.test.js b/apps/agent_frontend_a2a/tests/client.test.js new file mode 100644 index 00000000..e9c1cd5b --- /dev/null +++ b/apps/agent_frontend_a2a/tests/client.test.js @@ -0,0 +1,86 @@ +import test from 'node:test'; +import assert from 'node:assert/strict'; +import { A2AClient, TaskView, displayResponse, parseSSE, serviceURLs, selectVersion, normalizeResult, isStopped, needsInput, RpcError } from '../a2a-client.js'; + +function bodyFor(text, size = 1) { + const bytes=new TextEncoder().encode(text); + return new ReadableStream({start(controller){for(let i=0;i{ + const text=': ping\r\nid: 1\r\nevent: message\r\ndata: {"text":\r\ndata: "Olá 👋"}\r\n\r\ndata: second\n\n'; + const frames=[];for await(const frame of parseSSE(bodyFor(text)))frames.push(frame); + assert.deepEqual(frames,['{"text":\n"Olá 👋"}','second']); +}); +test('SSE discards an unfinished frame for recovery by GetTask',async()=>{ + const frames=[];for await(const frame of parseSSE(bodyFor('data: complete\n\ndata: truncated')))frames.push(frame); + assert.deepEqual(frames,['complete']); +}); +test('service URLs accept a base or /a2a and reject embedded credentials',()=>{ + assert.equal(serviceURLs('http://localhost:8000/a2a/').rpc,'http://localhost:8000/a2a'); + assert.equal(serviceURLs('https://host/prefix').card,'https://host/prefix/.well-known/agent-card.json'); + for(const url of ['javascript:alert(1)','https://user:secret@host','https://host?token=secret'])assert.throws(()=>serviceURLs(url)); +}); +test('version selection uses JSONRPC interfaces and allows explicit legacy compatibility',()=>{ + assert.equal(selectVersion({supportedInterfaces:[{protocolBinding:'JSONRPC',protocolVersion:'1.0'}]}),'1.0'); + assert.equal(selectVersion({protocolVersion:'0.3.0'}),'0.3'); + assert.equal(selectVersion({},'0.3'),'0.3'); + assert.throws(()=>selectVersion({supportedInterfaces:[{protocolBinding:'GRPC',protocolVersion:'1.0'}]})); +}); +test('both versions normalize task, message and streaming events',()=>{ + assert.equal(normalizeResult({statusUpdate:{taskId:'t'}}).type,'status'); + assert.equal(normalizeResult({kind:'artifact-update',taskId:'t'}).type,'artifact'); + assert.equal(normalizeResult({message:{messageId:'m'}}).type,'message'); + assert.throws(()=>normalizeResult({unexpected:true})); +}); +test('TaskView updates status, replaces snapshots and appends artifact chunks',()=>{ + const view=new TaskView({id:'t',artifacts:[]}); + view.apply({type:'artifact',value:{taskId:'t',artifact:{artifactId:'a',parts:[{text:'Hel'}]}}}); + view.apply({type:'artifact',value:{taskId:'t',append:true,artifact:{artifactId:'a',parts:[{text:'lo'}]}}}); + assert.equal(displayResponse(view.task).text,'Hello'); + view.apply({type:'status',value:{taskId:'t',contextId:'c',status:{state:'TASK_STATE_COMPLETED'}}}); + assert.ok(isStopped(view.task.status.state));assert.ok(needsInput('input-required')); + assert.throws(()=>view.apply({type:'status',value:{taskId:'other',status:{state:'working'}}})); +}); +test('framework text/data duplicates render once and native agent metadata survives',()=>{ + const native={text:'answer',metadata:{agent_responses:[{agent:'billing',text:'answer'}]}}; + const task={artifacts:[{parts:[{text:'answer'},{data:native}]}],status:{message:{parts:[{text:'answer'}]}}}; + assert.deepEqual(displayResponse(task),native); + const wrapped={parts:[{data:{text:'answer',metadata:{a2a_result:{task}}}}]}; + assert.deepEqual(displayResponse(wrapped),native); +}); +for(const legacy of [false,true]){ + test(`send/get/cancel use correct envelopes and credentials (${legacy?'0.3':'1.0'})`,async()=>{ + const requests=[]; + const fetchImpl=async function(url,options){ + assert.equal(this,globalThis); + const body=JSON.parse(options.body);requests.push({url,options,body}); + return Response.json({jsonrpc:'2.0',id:body.id,result:body.method.includes('send')||body.method==='SendMessage' ? (legacy?{kind:'task',id:'t',status:{state:'working'}}:{task:{id:'t',status:{state:'TASK_STATE_WORKING'}}}):{id:'t'}}); + }; + const client=new A2AClient({url:'https://service',version:legacy?'0.3':'1.0',headers:{Authorization:'Bearer test'},fetchImpl}); + await client.sendMessage('hello',{contextId:'c',agentId:'a',backendId:'billing',tenantId:'tenant',data:{business_context:{customer_key:'customer'}}}); + await client.getTask('t');await client.cancelTask('t'); + assert.deepEqual(requests.map(r=>r.body.method),legacy?['message/send','tasks/get','tasks/cancel']:['SendMessage','GetTask','CancelTask']); + const params=requests[0].body.params; + assert.equal(params.message.role,legacy?'user':'ROLE_USER');assert.equal(params.message.metadata.backend_id,'billing'); + assert.equal(params.message.metadata.agent_id,'a');assert.equal(params.message.contextId,'c'); + assert.equal(legacy?params.configuration.blocking:params.configuration.returnImmediately,!legacy); + assert.equal(requests[0].options.headers.Authorization,'Bearer test'); + assert.equal(requests[0].options.credentials,'omit'); + }); +} +test('JSON-RPC errors and mismatched response IDs never become successful responses',async()=>{ + const bad=new A2AClient({url:'https://service',fetchImpl:async()=>Response.json({jsonrpc:'2.0',id:'wrong',result:{}})}); + await assert.rejects(bad.getTask('t'),/JSON-RPC/); + const failure=new A2AClient({url:'https://service',fetchImpl:async(url,o)=>Response.json({jsonrpc:'2.0',id:JSON.parse(o.body).id,error:{code:-32001,message:'Task missing'}})}); + await assert.rejects(failure.getTask('t'),RpcError); +}); +test('POST SSE supports auth headers, validates envelopes and streams canonical events',async()=>{ + const client=new A2AClient({url:'https://service',headers:{'x-api-key':'test'},fetchImpl:async(url,o)=>{ + assert.equal(o.method,'POST');assert.equal(o.headers['x-api-key'],'test'); + const body=JSON.parse(o.body);assert.equal(body.method,'SubscribeToTask'); + return new Response(bodyFor(`data: ${JSON.stringify({jsonrpc:'2.0',id:body.id,result:{statusUpdate:{taskId:'t',status:{state:'TASK_STATE_COMPLETED'}}}})}\n\n`),{headers:{'content-type':'text/event-stream'}}); + }}); + const events=[];for await(const event of client.subscribe('t'))events.push(event); + assert.equal(events[0].type,'status');assert.equal(events[0].value.taskId,'t'); +}); diff --git a/apps/agent_frontend_a2a/tests/integration.test.js b/apps/agent_frontend_a2a/tests/integration.test.js new file mode 100644 index 00000000..f8614869 --- /dev/null +++ b/apps/agent_frontend_a2a/tests/integration.test.js @@ -0,0 +1,32 @@ +// Run against examples/a2a_frontend_demo.py with A2A_DEMO_URL=http://127.0.0.1:8765. +import test from 'node:test'; +import assert from 'node:assert/strict'; +import { A2AClient, TaskView, displayResponse, isStopped, stateName } from '../a2a-client.js'; +const url=process.env.A2A_DEMO_URL; +const delay=ms=>new Promise(resolve=>setTimeout(resolve,ms)); +async function complete(client,task){ + const deadline=Date.now()+6000; + while(!isStopped(task.status?.state)&&Date.now(){ + const client=new A2AClient({url,version});const card=await client.discover();assert.ok(card.capabilities.streaming); + const first=await client.sendMessage('hello',{contextId:crypto.randomUUID()});assert.equal(first.type,'task');assert.ok(!isStopped(first.value.status.state)); + const view=new TaskView(first.value);let count=0; + for await(const event of client.subscribe(first.value.id)){view.apply(event);count++;} + assert.ok(count>0);const final=await client.getTask(first.value.id); + assert.equal(stateName(final.status.state),'completed');assert.match(displayResponse(final).text,/hello/); + const prompt=await client.sendMessage('/confirmar',{contextId:crypto.randomUUID()}); + const waiting=await complete(client,prompt.value);assert.equal(stateName(waiting.status.state),'input-required'); + const resumed=await client.sendMessage('sim',{contextId:waiting.contextId,taskId:waiting.id}); + const finished=await complete(client,resumed.value);assert.equal(finished.id,waiting.id);assert.equal(stateName(finished.status.state),'completed'); + }); +} +test('SDK integration: cancellation and failed tasks',{skip:!url},async()=>{ + const client=new A2AClient({url}); + const slow=await client.sendMessage('/lento',{contextId:crypto.randomUUID()}); + await delay(50);const canceled=await client.cancelTask(slow.value.id);assert.equal(stateName(canceled.status.state),'canceled'); + const failed=await client.sendMessage('/erro',{contextId:crypto.randomUUID()}); + assert.equal(stateName((await complete(client,failed.value)).status.state),'failed'); +}); diff --git a/apps/agent_gateway/Dockerfile.a2a b/apps/agent_gateway/Dockerfile.a2a new file mode 100644 index 00000000..fc6a385a --- /dev/null +++ b/apps/agent_gateway/Dockerfile.a2a @@ -0,0 +1,7 @@ +FROM python:3.12-slim +WORKDIR /app +COPY libs/agent_framework /agent_framework +COPY apps/agent_gateway /app +RUN pip install --no-cache-dir -e '/agent_framework[a2a]' -r requirements.txt +ENV ENABLE_A2A=true +CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8010"] diff --git a/apps/agent_gateway/README.md b/apps/agent_gateway/README.md index 6ba217c1..0abf779f 100644 --- a/apps/agent_gateway/README.md +++ b/apps/agent_gateway/README.md @@ -71,3 +71,13 @@ O Gateway emite eventos de observabilidade: - `IC.GLOBAL_GATEWAY_COMPLETED` - `NOC.005` em falhas - `NOC.006` em conclusão HTTP + +## A2A opcional + +A fachada A2A 1.0, a compatibilidade 0.3 e o catálogo misto de backends são +explicados em [Integração A2A](../../docs/A2A_IMPLEMENTACAO.md). +A ativação é opcional (`ENABLE_A2A=false` por padrão). + +A fachada também oferece [Tasks A2A assíncronas e streaming](../../docs/A2A_ASSINCRONO.md), +com envio imediato, consulta, reconexão e cancelamento. O template de backend +pode publicar o mesmo protocolo diretamente. diff --git a/apps/agent_gateway/app/main.py b/apps/agent_gateway/app/main.py index 06c5cd81..9556ba7e 100644 --- a/apps/agent_gateway/app/main.py +++ b/apps/agent_gateway/app/main.py @@ -219,6 +219,10 @@ import asyncio @app.get("/gateway/events/{session_id:path}") async def gateway_events(session_id: str): + existing = await session_store.get(session_id) + if existing and existing.active_backend and registry.get(existing.active_backend).protocol == "a2a": + raise HTTPException(status_code=409, detail="A2A backend: use message/send; native event proxy is unavailable") + async def stream(): yield ( "event: connected\n" @@ -250,6 +254,9 @@ async def gateway_events(session_id: str): return backend = registry.get(backend_id) + if backend.protocol == "a2a": + yield 'event: error\ndata: {"message":"Native event proxy is unavailable for A2A backends"}\n\n' + return backend_base_url = ( getattr(backend, "base_url", None) @@ -325,4 +332,17 @@ async def gateway_events(session_id: str): "Connection": "keep-alive", "X-Accel-Buffering": "no", }, - ) \ No newline at end of file + ) + +# Opt-in: existing deployments do not import or require the A2A SDK. +if settings.ENABLE_A2A: + from agent_framework.a2a.server import install_a2a + + async def _a2a_gateway_handler(data: dict): + return await gateway_message(GatewayRequest(**data)) + + install_a2a( + app, _a2a_gateway_handler, registry, settings.A2A_PUBLIC_URL, + settings.A2A_SECURITY_SCHEMES or None, settings.A2A_SECURITY or None, + task_mode=settings.A2A_TASK_MODE, timeout_seconds=settings.A2A_TASK_TIMEOUT_SECONDS, + ) diff --git a/apps/agent_gateway/app/settings.py b/apps/agent_gateway/app/settings.py index 904644d7..77957e73 100644 --- a/apps/agent_gateway/app/settings.py +++ b/apps/agent_gateway/app/settings.py @@ -3,6 +3,7 @@ from __future__ import annotations from functools import lru_cache from typing import Literal +from pydantic import Field from pydantic_settings import BaseSettings, SettingsConfigDict @@ -16,6 +17,13 @@ class GatewaySettings(BaseSettings): API_PORT: int = 8010 CORS_ORIGINS: str = "http://localhost:5173" + ENABLE_A2A: bool = False + A2A_TASK_MODE: bool = True + A2A_TASK_TIMEOUT_SECONDS: float = Field(default=120.0, gt=0) + A2A_PUBLIC_URL: str = "http://localhost:8010" + A2A_SECURITY_SCHEMES: dict = Field(default_factory=dict) + A2A_SECURITY: list[dict[str, list[str]]] = Field(default_factory=list) + BACKENDS_CONFIG_PATH: str = "./config/backends.yaml" GLOBAL_ROUTING_MODE: Literal["router", "supervisor", "hybrid"] = "hybrid" GLOBAL_KEEP_ACTIVE_BACKEND: bool = True diff --git a/apps/agent_gateway/config/backends.a2a.example.yaml b/apps/agent_gateway/config/backends.a2a.example.yaml new file mode 100644 index 00000000..456785c1 --- /dev/null +++ b/apps/agent_gateway/config/backends.a2a.example.yaml @@ -0,0 +1,27 @@ +#default_backend: native +#backends: +# native: +# url: http://localhost:8000 +# description: Agente nativo do framework +# keywords: [conta, fatura] +# # protocol omitido equivale a native +# external: +# protocol: a2a +# a2a_version: '1.0' # ou '0.3' +# url: https://external-agent.example +# a2a_rpc_path: /a2a # endpoint definido pelo operador a partir do Agent Card +# a2a_agent_card_path: /.well-known/agent-card.json +# description: Agente externo A2A +# keywords: [suporte] +# a2a_headers_env: +# Authorization: EXTERNAL_AGENT_AUTHORIZATION # valor completo: Bearer + +default_backend: agent_template + +backends: + agent_template: + protocol: a2a + a2a_version: "1.0" + url: http://localhost:8000 + a2a_rpc_path: /a2a + a2a_agent_card_path: /.well-known/agent-card.json \ No newline at end of file diff --git a/docs/A2A_ASSINCRONO.md b/docs/A2A_ASSINCRONO.md new file mode 100644 index 00000000..eefd5607 --- /dev/null +++ b/docs/A2A_ASSINCRONO.md @@ -0,0 +1,248 @@ +# Agentes do framework com A2A assíncrono + +## O que foi implementado + +O agente criado a partir de `templates/agent_template_backend` pode expor o +protocolo nativo e A2A no mesmo processo. O Agent Gateway também oferece a +mesma fachada A2A. As duas fachadas usam o SDK oficial e suas operações de +Tasks/SSE, com interoperabilidade A2A 1.0 e 0.3. + +```text +Consumidor A2A ── Gateway A2A (Task) + ├── agente nativo (/gateway/message) + └── agente framework A2A (Task remota) + +Consumidor A2A ── agente framework A2A diretamente +``` + +O envio imediato devolve uma Task antes de terminar a operação de negócio. +A execução continua após a resposta HTTP e também após a desconexão do SSE. +O consumidor pode consultar o resultado ou se reconectar à tarefa em andamento. +Os endpoints nativos conservam seu comportamento; `/gateway/message` continua +aguardando o resultado, e o SSE nativo permanece separado do SSE A2A. + +## Habilitar no agente construído com o framework + +Na raiz do framework, instale a dependência opcional: + +```sh +python -m pip install -e 'libs/agent_framework[a2a]' -r templates/agent_template_backend/requirements.txt +``` + +No ambiente do backend: + +```dotenv +ENABLE_A2A=true +A2A_TASK_MODE=true +A2A_PUBLIC_URL=http://localhost:8000 +A2A_TASK_TIMEOUT_SECONDS=240 +``` + +Inicie o backend com suas configurações normais. Por exemplo, após preparar o +`.env` e os arquivos de configuração do agente: + +```sh +cd templates/agent_template_backend +python -m uvicorn app.main:app --host 0.0.0.0 --port 8000 +``` + +O endereço público real deve ser usado em `A2A_PUBLIC_URL`. A configuração de +exemplo está em `.env.a2a.example`. Há também uma imagem opcional, construída +na raiz do framework: + +```sh +docker build -f templates/agent_template_backend/Dockerfile.a2a -t framework-agent-a2a . +``` + +Os perfis de `agents.yaml` são publicados como skills no Agent Card: +`GET /.well-known/agent-card.json`. Use `message.metadata.agent_id` para escolher +um perfil; sem ele, o backend mantém sua seleção padrão. + +## Registrar esse agente no Agent Gateway + +```yaml +backends: + pedidos_a2a: + protocol: a2a + a2a_version: '1.0' + url: http://backend-pedidos:8000 + a2a_rpc_path: /a2a + description: Consulta e acompanhamento de pedidos + keywords: [pedido, entrega, rastreamento] + default_agent_id: pedidos +``` + +No gateway, habilite `ENABLE_A2A=true`, `A2A_TASK_MODE=true` e configure seu próprio +`A2A_PUBLIC_URL`. Para uma execução mais longa, ajuste os limites, por exemplo: +`A2A_TASK_TIMEOUT_SECONDS=300` e `BACKEND_TIMEOUT_SECONDS=270`, com o timeout do +backend em 240 segundos. A duração efetiva é limitada pelo menor prazo aplicável. + +O gateway devolve sua própria Task imediatamente e consulta a Task remota até +conclusão ou solicitação de input/autenticação. Eventos de status remoto são +publicados como `a2a.backend.status`. Os IDs local e remoto podem ser distintos. +O resultado nativo inclui `metadata.a2a_task_id` e `metadata.a2a_context_id` para +correlação; o consumidor externo continua acompanhando o ID emitido pelo gateway. + +## Envio assíncrono A2A 1.0 + +Envie um `POST /a2a` com `Content-Type: application/json` e +`A2A-Version: 1.0`: + +```json +{ + "jsonrpc": "2.0", + "id": "request-1", + "method": "SendMessage", + "params": { + "configuration": {"returnImmediately": true}, + "message": { + "messageId": "message-1", + "contextId": "conversation-123", + "role": "ROLE_USER", + "parts": [{"text": "Consulte meu pedido"}], + "metadata": {"agent_id": "pedidos", "tenant_id": "default"} + } + } +} +``` + +A resposta contém `result.task.id`, `contextId` e um estado inicial como +`TASK_STATE_SUBMITTED` ou `TASK_STATE_WORKING`. Guarde esse ID e consulte: + +```json +{ + "jsonrpc": "2.0", + "id": "request-2", + "method": "GetTask", + "params": {"id": "ID_DA_TAREFA"} +} +``` + +O resultado está em `result`. Quando `status.state` for `TASK_STATE_COMPLETED`, +o conteúdo final estará em `artifacts[].parts`. Há uma parte de texto e uma +parte `data` com a resposta original do framework. + +Sem `returnImmediately: true`, o envio aguarda o estado final ou interrompido, +retornando uma Task no mesmo formato. Para manter o comportamento da primeira +entrega A2A, que retornava Message, configure `A2A_TASK_MODE=false`; nesse modo, +streaming fica desabilitado e não há Tasks locais. + +## Streaming e reconexão + +Para enviar e acompanhar, use `SendStreamingMessage` no lugar de `SendMessage` +com os mesmos parâmetros. A resposta tem `Content-Type: text/event-stream`. +Cada campo `data:` contém uma resposta JSON-RPC. Os eventos contêm: + +1. A Task inicial. +2. Atualizações de status: submitted/working e eventos de progresso. +3. O artefato de resposta. +4. O estado completed, failed, canceled, input-required ou auth-required. + +O backend do framework publica marcos como `flow.start`, `session.upserted`, +`workflow.started`, `workflow.completed` e `message.responded`, quando o fluxo +passa por essas etapas. Eles aparecem em +`result.statusUpdate.metadata.framework_event`, com os dados correspondentes +em `framework_payload`. Isso não implica transmissão de tokens do LLM. + +Para se reconectar a uma Task em andamento: + +```json +{ + "jsonrpc": "2.0", + "id": "request-3", + "method": "SubscribeToTask", + "params": {"id": "ID_DA_TAREFA"} +} +``` + +A reconexão entrega o estado atual e os eventos seguintes; não há promessa de +replay de todos os eventos perdidos. Se a tarefa já terminou, consulte `GetTask`. + +## Cancelamento, falhas e continuidade + +`CancelTask`, com `params.id`, cancela a execução local em andamento. O SDK +interrompe a coroutine e registra `TASK_STATE_CANCELED`. Isso não desfaz efeitos +já confirmados em ferramentas ou sistemas externos. Quando o gateway está +consultando um agente remoto, parar a tarefa local não garante cancelar a +operação que já foi aceita pelo serviço remoto. + +Exceções e expiração de `A2A_TASK_TIMEOUT_SECONDS` resultam em +`TASK_STATE_FAILED`, consultável depois, com mensagem genérica. Detalhes internos +permanecem nos logs do servidor. + +Para uma interação que exige nova entrada, um handler pode retornar +`metadata.a2a_task_state: "input-required"` ou `"auth-required"`. O consumidor +envia outra mensagem com o mesmo `taskId` e `contextId`. A tarefa é retomada; +uma tarefa concluída não pode ser reexecutada. O adaptador de saída preserva a +continuidade de tarefas remotas nesses estados. + +Os workflows nativos que não declaram esse estado concluem uma Task por turno; +a conversa continua com uma nova Task e o mesmo `contextId`, usando as sessões +e memórias já existentes no framework. + +## Equivalentes A2A 0.3 + +Use `A2A-Version: 0.3`, `role: "user"` e os discriminadores `kind` de 0.3. + +| A2A 1.0 | A2A 0.3 | +| --- | --- | +| `SendMessage` + `returnImmediately: true` | `message/send` + `blocking: false` | +| `SendStreamingMessage` | `message/stream` | +| `GetTask` | `tasks/get` | +| `SubscribeToTask` | `tasks/resubscribe` | +| `CancelTask` | `tasks/cancel` | + +Em 0.3, `result` contém diretamente a Task, sem o envelope `task`. Os estados +usam `submitted`, `working`, `completed`, `failed`, etc. + +## Armazenamento e autenticação + +O padrão usa o TaskStore em memória do SDK. Tarefas e resultados são perdidos +quando o processo reinicia. Para múltiplos workers/réplicas, use afinidade de +requisições ao mesmo processo durante a execução ou configure armazenamento e +coordenação compartilhados. Um TaskStore persistente, sozinho, não torna a +execução retomável após falha. As funções `install_a2a` e `install_backend_a2a` +aceitam `task_store` para integração com um armazenamento do SDK. + +O backend opcional reutiliza `AGENT_AUTH_*`; o gateway usa +`AGENT_GATEWAY_AUTH_*`. A identidade autenticada pelo middleware define o dono +das Tasks, incluindo consultas e cancelamento. Usuários anônimos compartilham +o escopo anônimo. `tenant_id` em metadata continua sendo dado do pedido, não +credencial nem isolamento de autorização. Clientes que compartilham uma mesma +credencial/principal compartilham o escopo de tarefas. + +`A2A_SECURITY_SCHEMES` e `A2A_SECURITY` documentam os mecanismos no Agent Card, +como descrito no guia geral. SSE está habilitado no modo de Tasks; +notificações push e gRPC continuam fora desta implementação. + +## Validação + +A suíte inclui envio imediato enquanto o handler permanece bloqueado, polling, +streaming com progresso e resultado, cancelamento, falhas, timeout, retomada, +separação entre identidades autenticadas e gateway chamando um backend A2A real +em aplicação local. Há também testes HTTP em `127.0.0.1` que verificam a chegada +do primeiro evento antes da conclusão e reconexão após desconexão. + +```sh +python -m pip install pytest pytest-asyncio uvicorn +PYTHONPATH=libs/agent_framework/src pytest \ + tests/test_a2a_interoperability.py tests/test_a2a_async_tasks.py \ + tests/test_a2a_http_streaming.py \ + tests/unit/test_sse.py tests/unit/test_sse_replay_dedup.py \ + tests/unit/test_authentication.py tests/unit/test_authentication_policies.py -q +``` + +Os testes não acessam OCI nem serviços de negócio externos. Em ambientes sem +permissão para abrir portas locais, execute o teste HTTP em um ambiente que +permita bind em loopback. + +Especificação oficial: https://a2a-protocol.org/latest/specification/ +SDK oficial: https://github.com/a2aproject/a2a-python + +## Frontend pronto para A2A + +A interface `apps/agent_frontend_a2a` consome este protocolo diretamente, com +Tasks, streaming, consulta e cancelamento. Consulte +[Frontend A2A](../apps/agent_frontend_a2a/README.md) para a demonstração sem OCI +e os comandos de execução com o backend ou Gateway. O frontend nativo continua +em `apps/agent_frontend`. diff --git a/docs/A2A_IMPLEMENTACAO.md b/docs/A2A_IMPLEMENTACAO.md new file mode 100644 index 00000000..d96b3eec --- /dev/null +++ b/docs/A2A_IMPLEMENTACAO.md @@ -0,0 +1,145 @@ +# Integração A2A no Agent Framework OCI + +## Decisão de arquitetura + +O Agent Gateway existente passa a ter uma fachada A2A opcional. Não é necessário +manter um segundo roteador, duplicar o supervisor ou migrar os agentes atuais. + +```text +Consumidor atual ── /gateway/message ─┐ + ├─ Agent Gateway / supervisor ─┬─ backend native +Consumidor A2A ──── /a2a ─────────────┘ └─ backend a2a +``` + +O backend nativo continua recebendo `{channel, payload, agent_id, tenant_id}`. +A ausência de `protocol` no catálogo mantém `native`. A2A fica desativado por +padrão; o SDK só é importado ao habilitar a fachada ou selecionar um backend A2A. +Não foi alterado o runtime dos agentes, os workflows, os canais ou os MCPs. + +A entrada usa A2A **1.0 / JSON-RPC**, SDK oficial `a2a-sdk==1.2.2`, com adaptação +0.3 disponibilizada pelo próprio SDK. Agentes externos podem ser registrados +com `a2a_version: '1.0'` ou `'0.3'`; não há detecção heurística de versão. + +## Ativação local + +Na raiz do framework: + +```sh +python -m pip install -e 'libs/agent_framework[a2a]' -r apps/agent_gateway/requirements.txt +export ENABLE_A2A=true +export A2A_PUBLIC_URL=https://meu-gateway.example +export BACKENDS_CONFIG_PATH="$PWD/apps/agent_gateway/config/backends.a2a.example.yaml" +python -m uvicorn app.main:app --app-dir apps/agent_gateway --port 8010 +``` + +Edite primeiro as URLs do catálogo de exemplo. Para agentes externos, informe o +endpoint JSON-RPC anunciado no Agent Card como `url` + `a2a_rpc_path`. +Credenciais ficam em variáveis de ambiente referenciadas por `a2a_headers_env`. +Os valores não são publicados em `/backends`. + +A imagem opcional pode ser construída a partir da raiz: + +```sh +docker build -f apps/agent_gateway/Dockerfile.a2a -t agent-gateway-a2a . +``` + +Defina `A2A_PUBLIC_URL` no ambiente do container e monte o catálogo de backends. +A imagem e os comandos anteriores de implantação continuam disponíveis. + +## Descoberta e chamada + +O Agent Card fica em `GET /.well-known/agent-card.json`; as skills são derivadas +dos backends habilitados. As mensagens chegam em `POST /a2a`. + +Exemplo A2A 1.0, com cabeçalho `A2A-Version: 1.0`: + +```json +{ + "jsonrpc": "2.0", + "id": "request-1", + "method": "SendMessage", + "params": { + "message": { + "messageId": "message-1", + "contextId": "conversation-123", + "role": "ROLE_USER", + "parts": [ + {"text": "Quero consultar minha fatura"}, + {"data": {"customer_id": "123"}} + ], + "metadata": {"tenant_id": "default"} + } + } +} +``` + +No modo de tarefas (padrão), a resposta fica em `result.task`. O resultado de +negócio está em `artifacts[].parts`, com texto e uma parte `data` contendo a +resposta do framework. Com `A2A_TASK_MODE=false`, permanece em `result.message`. Reutilize `contextId` nos próximos turnos. No modo 0.3, +use o cabeçalho `A2A-Version: 0.3`, método `message/send`, role `user` e os +campos `kind` da especificação 0.3; o resultado é uma Task diretamente +(ou Message com `A2A_TASK_MODE=false`). + +Partes `data` devem conter objetos e são incorporadas ao payload de negócio. +A identidade da sessão e o texto do protocolo prevalecem sobre esses dados. +Arquivos recebidos pela fachada são rejeitados explicitamente. + +## Conversas, tarefas e limites + +- A fachada cria **Tasks A2A** e permite envio imediato, consulta, cancelamento + e streaming SSE. `A2A_TASK_MODE=false` mantém o modo anterior de Messages. + Consulte [A2A assíncrono](A2A_ASSINCRONO.md) para os exemplos completos. +- O adaptador de saída aceita Message ou Task; consulta `GetTask`/`tasks/get` + enquanto o agente está trabalhando, dentro de um prazo total configurável + por `BACKEND_TIMEOUT_SECONDS`. Não repete automaticamente operações de envio. +- Tarefas que pedem input/autenticação preservam estado e identidade para o + próximo turno. Tarefas concluídas não são reutilizadas. +- A continuidade remota é guardada no processo, por tenant/backend/sessão, + com validade de uma hora e limite de 10 mil entradas. Em múltiplas réplicas, + use afinidade de sessão ou implemente armazenamento compartilhado antes de + depender dessa continuidade. Reiniciar o processo limpa esse cache. +- `metadata.a2a_result` preserva o resultado remoto, incluindo dados/artefatos; + `a2a_context_id`, `a2a_task_id` e `a2a_state` permitem inspeção e continuidade + explícita no payload quando necessário. +- Streaming SSE está disponível no modo de tarefas. Push notifications e gRPC + não são oferecidos. O proxy de eventos nativo não é utilizado para agentes + A2A: use `SendStreamingMessage`/`SubscribeToTask` e seus equivalentes 0.3. +- O template `templates/agent_template_backend` também expõe A2A diretamente + quando `ENABLE_A2A=true`, sem passar pelo gateway global. + +## Autenticação + +As rotas A2A compartilham o middleware de autenticação já instalado no gateway +(`AGENT_GATEWAY_AUTH_*`). Habilitar A2A não habilita autenticação automaticamente. +O tenant em metadata mantém a semântica atual do framework: é dado do pedido, +não uma autorização concedida pelo protocolo. Políticas de acesso a tenants +continuam sendo responsabilidade da configuração de autenticação/autorização. + +Quando houver autenticação, declare-a também no Agent Card, por exemplo: + +```sh +export A2A_SECURITY_SCHEMES='{"gatewayBearer":{"httpAuthSecurityScheme":{"scheme":"bearer"}}}' +export A2A_SECURITY='[{"gatewayBearer":[]}]' +``` + +Essas declarações documentam o mecanismo; quem verifica as credenciais é o +middleware. Se desejar descoberta pública, adicione somente o caminho do Agent +Card à configuração de caminhos públicos; `/a2a` permanece protegido. + +## Validação + +Testes em `tests/test_a2a_interoperability.py` cobrem os dois formatos de entrada, +o contrato nativo, os dois formatos de saída, polling, erros JSON-RPC, +continuidade de tarefas, separação entre tenants e limite de tempo. + +```sh +PYTHONPATH=libs/agent_framework/src pytest tests/test_a2a_interoperability.py -q +``` + +Validação com transportes HTTP simulados e aplicação FastAPI local; não substitui +um teste de integração com os serviços OCI e agentes externos da implantação. + +Referências oficiais: +- https://a2a-protocol.org/latest/specification/ +- https://a2a-protocol.org/v0.3.0/specification/ +- https://github.com/a2aproject/a2a-python diff --git a/examples/a2a_frontend_demo.py b/examples/a2a_frontend_demo.py new file mode 100644 index 00000000..12e736fd --- /dev/null +++ b/examples/a2a_frontend_demo.py @@ -0,0 +1,43 @@ +"""Local frontend demo, using the actual framework A2A adapter and SDK. + +From the framework root: + pip install -e 'libs/agent_framework[a2a]' uvicorn + uvicorn examples.a2a_frontend_demo:app --host 127.0.0.1 --port 8000 + +This demo simulates business processing; it never calls OCI or an external tool. +""" +import asyncio +import os +from pathlib import Path +from types import SimpleNamespace + +from fastapi import FastAPI +from fastapi.staticfiles import StaticFiles +from agent_framework.a2a.server import install_backend_a2a +from agent_framework.a2a.progress import report_progress + +app = FastAPI(title='Agent Framework · A2A Frontend Demo') + + +async def demo_agent(request): + text = request['payload']['message'] + await report_progress('workflow.started', {'agent': request.get('agent_id') or 'demo'}) + await asyncio.sleep(1) + if text.strip().lower() == '/erro': + raise RuntimeError('Simulated demo error') + if text.strip().lower() == '/lento': + await asyncio.sleep(20) + await report_progress('workflow.completed', {'agent': 'demo'}) + if text.strip().lower() == '/confirmar': + return {'text': 'Deseja continuar com a demonstração? Responda sim para prosseguir.', + 'metadata': {'a2a_task_state': 'input-required', 'agent_id': 'demo'}} + return {'text': f'Recebi sua mensagem: “{text}”.\n\nA tarefa A2A foi concluída. Este exemplo usa o adaptador real do framework, com processamento de negócio simulado.', + 'session_id': request['session_id'], 'metadata': {'agent_id': 'demo'}} + + +profiles = SimpleNamespace(list_profiles=lambda: [SimpleNamespace( + agent_id='demo', name='Agente de demonstração', description='Exemplo local de agente A2A assíncrono')]) +settings = SimpleNamespace(A2A_PUBLIC_URL=os.getenv('A2A_PUBLIC_URL', 'http://localhost:8000'), + A2A_TASK_MODE=True, A2A_TASK_TIMEOUT_SECONDS=60, A2A_SECURITY_SCHEMES={}, A2A_SECURITY=[], APP_NAME='Agente de demonstração') +install_backend_a2a(app, demo_agent, profiles, settings) +app.mount('/', StaticFiles(directory=Path(__file__).resolve().parents[1] / 'apps' / 'agent_frontend_a2a', html=True), name='frontend') diff --git a/libs/agent_framework/pyproject.toml b/libs/agent_framework/pyproject.toml index e5ffbfa5..961f83a1 100644 --- a/libs/agent_framework/pyproject.toml +++ b/libs/agent_framework/pyproject.toml @@ -27,6 +27,9 @@ dependencies = [ "PyJWT[crypto]>=2.9.0" ] +[project.optional-dependencies] +a2a = ["a2a-sdk[http-server]==1.2.2"] + [tool.setuptools.packages.find] where = ["src"] diff --git a/libs/agent_framework/src/agent_framework/a2a/__init__.py b/libs/agent_framework/src/agent_framework/a2a/__init__.py new file mode 100644 index 00000000..dc8ffd3a --- /dev/null +++ b/libs/agent_framework/src/agent_framework/a2a/__init__.py @@ -0,0 +1 @@ +"""Optional A2A interoperability; imported only when enabled.""" diff --git a/libs/agent_framework/src/agent_framework/a2a/client.py b/libs/agent_framework/src/agent_framework/a2a/client.py new file mode 100644 index 00000000..daf50c49 --- /dev/null +++ b/libs/agent_framework/src/agent_framework/a2a/client.py @@ -0,0 +1,108 @@ +"""A2A JSON-RPC adapter. URLs and credential references are operator configured.""" +import asyncio +import json +import os +from uuid import uuid4, uuid5, NAMESPACE_URL + +import httpx +from google.protobuf.json_format import ParseDict, MessageToDict +from a2a.types import SendMessageResponse, Task + +from .progress import report_progress + + +def _normalize_legacy(result): + # Preserve the original wire response separately; normalize only presentation. + if result.get('kind') == 'message': + return {'message': result} + if result.get('kind') == 'task': + return {'task': result} + raise ValueError('Expected an A2A Message or Task') + + +async def call_a2a(backend, request, timeout_seconds): + payload = request.get('payload') or {} + session = payload.get('session_id') or str(uuid4()) + tenant = request.get('tenant_id') or 'default' + context_id = str(uuid5(NAMESPACE_URL, json.dumps([tenant, backend.backend_id, session]))) + legacy = backend.a2a_version == '0.3' + text = str(payload.get('message') or payload.get('text') or payload.get('transcript') or payload.get('content') or '') + parts = [{'text': text}, {'data': payload}] + message = {'role': 'ROLE_USER', 'messageId': str(uuid4()), + 'contextId': payload.get('a2a_context_id') or context_id, 'parts': parts, + 'metadata': {'tenant_id': tenant, 'agent_id': request.get('agent_id') or backend.default_agent_id}} + if payload.get('a2a_task_id'): + message['taskId'] = payload['a2a_task_id'] + if legacy: + message.update(kind='message', role='user') + for part in parts: + part['kind'] = 'text' if 'text' in part else 'data' + headers = {'A2A-Version': backend.a2a_version} + for header, env_name in backend.a2a_headers_env.items(): + value = os.environ.get(env_name) + if not value: + raise ValueError(f'Missing credential environment variable: {env_name}') + headers[header] = value + endpoint = backend.base_url + backend.a2a_rpc_path + + async with asyncio.timeout(timeout_seconds): + async with httpx.AsyncClient(timeout=timeout_seconds, headers=headers) as client: + async def rpc(method, params): + request_id = str(uuid4()) + response = await client.post(endpoint, json={ + 'jsonrpc': '2.0', 'id': request_id, 'method': method, 'params': params, + }) + response.raise_for_status() + data = response.json() + if data.get('jsonrpc') != '2.0' or data.get('id') != request_id: + raise ValueError('Invalid A2A JSON-RPC response') + if 'error' in data: + raise RuntimeError(f'A2A error {data["error"].get("code")}') + result = data.get('result') + if not isinstance(result, dict): + raise ValueError('Missing A2A result') + return result + + raw = await rpc('message/send' if legacy else 'SendMessage', { + 'message': message, + 'configuration': {'blocking': False} if legacy else {'returnImmediately': True}, + }) + result = _normalize_legacy(raw) if legacy else raw + if not legacy: + result = MessageToDict(ParseDict(result, SendMessageResponse())) + while 'task' in result: + task = result['task'] + state = task.get('status', {}).get('state', '') + await report_progress('a2a.backend.status', { + 'backend_id': backend.backend_id, 'task_id': task.get('id'), 'state': state, + }) + if state not in {'submitted', 'working', 'TASK_STATE_SUBMITTED', 'TASK_STATE_WORKING'}: + break + await asyncio.sleep(backend.a2a_poll_interval_seconds) + raw = await rpc('tasks/get' if legacy else 'GetTask', {'id': task['id']}) + result = {'task': raw if legacy else MessageToDict(ParseDict(raw, Task()))} + + task = result.get('task') + if task: + state, task_id = task['status']['state'], task['id'] + if state in {'failed', 'canceled', 'rejected', 'unknown', 'TASK_STATE_FAILED', 'TASK_STATE_CANCELED', 'TASK_STATE_REJECTED', 'TASK_STATE_UNSPECIFIED'}: + raise RuntimeError(f'A2A task ended with state: {state}') + output_parts = [p for a in task.get('artifacts', []) for p in a.get('parts', [])] + if not output_parts and task['status'].get('message'): + output_parts += task['status']['message'].get('parts', []) + if not output_parts: + messages = [m for m in task.get('history', []) if m.get('role') in {'agent', 'ROLE_AGENT'}] + output_parts = messages[-1].get('parts', []) if messages else [] + remote_context = task.get('contextId') + elif result.get('message'): + reply = result['message'] + output_parts = reply.get('parts', []) + state, task_id, remote_context = None, None, reply.get('contextId') + else: + raise ValueError('Expected an A2A Message or Task') + return { + 'channel': request.get('channel', 'web'), 'session_id': session, + 'text': '\n'.join(p['text'] for p in output_parts if 'text' in p), + 'metadata': {'a2a_context_id': remote_context, 'a2a_task_id': task_id, + 'a2a_state': state, 'a2a_result': raw}, + } diff --git a/libs/agent_framework/src/agent_framework/a2a/progress.py b/libs/agent_framework/src/agent_framework/a2a/progress.py new file mode 100644 index 00000000..c8b811aa --- /dev/null +++ b/libs/agent_framework/src/agent_framework/a2a/progress.py @@ -0,0 +1,10 @@ +"""Task-local progress hook; safe to import without the optional A2A SDK.""" +from contextvars import ContextVar + +_progress_callback = ContextVar('framework_a2a_progress', default=None) + + +async def report_progress(event: str, payload: dict | None = None): + callback = _progress_callback.get() + if callback is not None: + await callback(event, payload or {}) diff --git a/libs/agent_framework/src/agent_framework/a2a/server.py b/libs/agent_framework/src/agent_framework/a2a/server.py new file mode 100644 index 00000000..b6a162dc --- /dev/null +++ b/libs/agent_framework/src/agent_framework/a2a/server.py @@ -0,0 +1,181 @@ +"""A2A 1.0 Tasks and SSE with SDK-managed 0.3 compatibility.""" +import asyncio +import logging +from uuid import uuid4 + +from google.protobuf.json_format import MessageToDict, ParseDict +from a2a.auth.user import User +from a2a.server.agent_execution import AgentExecutor +from a2a.server.routes import ( + DefaultServerCallContextBuilder, add_a2a_routes_to_fastapi, + create_agent_card_routes, create_jsonrpc_routes, +) +from a2a.server.request_handlers import DefaultRequestHandler +from a2a.server.tasks import InMemoryTaskStore, TaskUpdater +from a2a.types import AgentCard, Message, Part, Role, Task, TaskState, TaskStatus +from a2a.utils.errors import InvalidParamsError, TaskNotCancelableError, UnsupportedOperationError + +from .progress import _progress_callback + +logger = logging.getLogger(__name__) + + +class _PrincipalUser(User): + def __init__(self, principal): + self.principal = principal + + @property + def is_authenticated(self): + return True + + @property + def user_name(self): + # Use middleware identity, never client-supplied message metadata. + return f'{self.principal.scheme}:{self.principal.subject}' + + +class FrameworkCallContextBuilder(DefaultServerCallContextBuilder): + def build_user(self, request): + principal = getattr(request.state, 'auth_principal', None) + return _PrincipalUser(principal) if principal else super().build_user(request) + + +def _response_message(response, context_id, task_id=None): + data = { + 'messageId': str(uuid4()), 'role': 'ROLE_AGENT', 'contextId': context_id, + 'parts': [{'text': str(response.get('text') or response.get('speak') or '')}, + {'data': response}], + } + if task_id: + data['taskId'] = task_id + return ParseDict(data, Message()) + + +class GatewayExecutor(AgentExecutor): + """Wrap any native async handler as an A2A task, without detached worker hacks.""" + + def __init__(self, handler, task_mode=True, timeout_seconds=120.0): + self.handler = handler + self.task_mode = task_mode + self.timeout_seconds = timeout_seconds + + async def execute(self, context, event_queue): + message = context.message + if message is None or message.role != Role.ROLE_USER: + raise InvalidParamsError('Expected a user message') + payload = {} + for part in MessageToDict(message).get('parts', []): + if 'data' in part: + if not isinstance(part['data'], dict): + raise InvalidParamsError('Structured data must be an object') + payload.update(part['data']) + elif 'text' not in part: + raise UnsupportedOperationError('Only text and structured data are supported') + metadata = MessageToDict(message.metadata) + session_id = context.context_id or str(uuid4()) + payload.update(message=context.get_user_input(), session_id=session_id) + request = { + 'channel': 'web', 'payload': payload, 'session_id': session_id, + 'tenant_id': metadata.get('tenant_id'), 'agent_id': metadata.get('agent_id'), + 'metadata': metadata, 'backend_id': metadata.get('backend_id'), + } + if not self.task_mode: + async with asyncio.timeout(self.timeout_seconds): + response = await self.handler(request) + await event_queue.enqueue_event(_response_message(response, session_id)) + return + + updater = TaskUpdater(event_queue, context.task_id, session_id) + if context.current_task is None: + await event_queue.enqueue_event(Task( + id=context.task_id, context_id=session_id, + status=TaskStatus(state=TaskState.TASK_STATE_SUBMITTED), + history=[message], + )) + await updater.update_status(TaskState.TASK_STATE_WORKING) + + async def progress(event, data): + await updater.update_status(TaskState.TASK_STATE_WORKING, metadata={ + 'framework_event': event, 'framework_payload': data, + }) + + token = _progress_callback.set(progress) + try: + async with asyncio.timeout(self.timeout_seconds): + response = await self.handler(request) + reply = _response_message(response, session_id, context.task_id) + response_metadata = response.get('metadata') or {} + requested_state = response_metadata.get('a2a_state') or response_metadata.get('a2a_task_state') + state = { + 'input-required': TaskState.TASK_STATE_INPUT_REQUIRED, + 'auth-required': TaskState.TASK_STATE_AUTH_REQUIRED, + 'TASK_STATE_INPUT_REQUIRED': TaskState.TASK_STATE_INPUT_REQUIRED, + 'TASK_STATE_AUTH_REQUIRED': TaskState.TASK_STATE_AUTH_REQUIRED, + }.get(requested_state, TaskState.TASK_STATE_COMPLETED) + await updater.add_artifact(list(reply.parts), artifact_id='response', name='Framework response', last_chunk=True) + await updater.update_status(state, message=reply) + except asyncio.CancelledError: + raise + except Exception as exc: + logger.exception('A2A task %s failed', context.task_id) + text = 'Processing timed out' if isinstance(exc, TimeoutError) else 'Agent processing failed' + await updater.update_status(TaskState.TASK_STATE_FAILED, message=_response_message({'text': text}, session_id, context.task_id)) + finally: + _progress_callback.reset(token) + + async def cancel(self, context, event_queue): + if not self.task_mode: + raise TaskNotCancelableError() + # The SDK cancels the running execute coroutine after this callback. + await TaskUpdater(event_queue, context.task_id, context.context_id).update_status(TaskState.TASK_STATE_CANCELED) + + +def install_a2a(app, handler, registry, public_url, security_schemes=None, security=None, + *, task_mode=True, timeout_seconds=120.0, task_store=None, + name='Agent Framework OCI Gateway', skills=None): + """Reuse parent middleware. A persistent SDK TaskStore may be injected.""" + if timeout_seconds <= 0: + raise ValueError('A2A timeout_seconds must be positive') + if skills is None: + skills = [{'id': b.backend_id, 'name': b.name or b.backend_id, + 'description': b.description or b.backend_id, + 'tags': b.domains or [b.backend_id], 'examples': b.examples} + for b in registry.list()] + card = ParseDict({ + 'name': name, 'version': '1.1.0', + 'description': 'Native framework agents and A2A interoperability', + 'supportedInterfaces': [{'url': public_url.rstrip('/') + '/a2a', + 'protocolBinding': 'JSONRPC', 'protocolVersion': '1.0'}], + 'capabilities': {'streaming': task_mode, 'pushNotifications': False}, + 'defaultInputModes': ['text/plain', 'application/json'], + 'defaultOutputModes': ['text/plain', 'application/json'], + 'securitySchemes': security_schemes or {}, + 'securityRequirements': [{'schemes': {k: {'list': v} for k, v in item.items()}} + for item in (security or [])], + 'skills': skills, + }, AgentCard()) + request_handler = DefaultRequestHandler( + GatewayExecutor(handler, task_mode=task_mode, timeout_seconds=timeout_seconds), + task_store if task_store is not None else InMemoryTaskStore(), card, + ) + add_a2a_routes_to_fastapi( + app, agent_card_routes=create_agent_card_routes(card), + jsonrpc_routes=create_jsonrpc_routes( + request_handler, rpc_url='/a2a', enable_v0_3_compat=True, + context_builder=FrameworkCallContextBuilder(), + ), + ) + return request_handler + + +def install_backend_a2a(app, handler, profiles, settings, *, task_store=None): + """Expose a framework backend directly, with skills built from agent profiles.""" + return install_a2a( + app, handler, None, settings.A2A_PUBLIC_URL, + settings.A2A_SECURITY_SCHEMES or None, settings.A2A_SECURITY or None, + task_mode=settings.A2A_TASK_MODE, timeout_seconds=settings.A2A_TASK_TIMEOUT_SECONDS, + task_store=task_store, name=settings.APP_NAME, + skills=[{'id': p.agent_id, 'name': p.name or p.agent_id, + 'description': p.description or p.agent_id, 'tags': [p.agent_id]} + for p in profiles.list_profiles()], + ) diff --git a/libs/agent_framework/src/agent_framework/config/settings.py b/libs/agent_framework/src/agent_framework/config/settings.py index 2225c282..72df0caf 100644 --- a/libs/agent_framework/src/agent_framework/config/settings.py +++ b/libs/agent_framework/src/agent_framework/config/settings.py @@ -14,6 +14,13 @@ load_dotenv(override=False) class Settings(BaseSettings): model_config = SettingsConfigDict(env_file='.env', env_file_encoding='utf-8', extra='ignore') + ENABLE_A2A: bool = False + A2A_PUBLIC_URL: str = 'http://localhost:8000' + A2A_TASK_MODE: bool = True + A2A_TASK_TIMEOUT_SECONDS: float = Field(default=120.0, gt=0) + A2A_SECURITY_SCHEMES: dict = Field(default_factory=dict) + A2A_SECURITY: list[dict[str, list[str]]] = Field(default_factory=list) + APP_NAME: str = 'ai-agent-template' APP_ENV: str = 'local' LOG_LEVEL: str = 'INFO' diff --git a/libs/agent_framework/src/agent_framework/global_supervisor/client.py b/libs/agent_framework/src/agent_framework/global_supervisor/client.py index 2fec58e9..9393ebe3 100644 --- a/libs/agent_framework/src/agent_framework/global_supervisor/client.py +++ b/libs/agent_framework/src/agent_framework/global_supervisor/client.py @@ -1,6 +1,8 @@ from __future__ import annotations import time +import os +from collections import OrderedDict from typing import Any import httpx @@ -11,6 +13,8 @@ from .models import BackendCallResult, BackendDefinition, GlobalRouteDecision class BackendClient: def __init__(self, timeout_seconds: float = 120.0): self.timeout_seconds = timeout_seconds + # Bounded, process-local continuation cache; native calls do not use it. + self._a2a_sessions = OrderedDict() async def call_message( self, @@ -19,6 +23,27 @@ class BackendClient: route_decision: GlobalRouteDecision, use_sse: bool = False, ) -> BackendCallResult: + if backend.protocol == "a2a": + from agent_framework.a2a.client import call_a2a + started = time.monotonic() + payload = dict(request_payload.get("payload") or {}) + key = (request_payload.get("tenant_id") or "default", backend.backend_id, payload.get("session_id")) + previous = self._a2a_sessions.get(key) if key[2] else None + if previous and time.monotonic() - previous[0] < 3600: + payload.setdefault("a2a_context_id", previous[1].get("a2a_context_id")) + if previous[1].get("a2a_state") in {"input-required", "auth-required", "TASK_STATE_INPUT_REQUIRED", "TASK_STATE_AUTH_REQUIRED"}: + payload.setdefault("a2a_task_id", previous[1].get("a2a_task_id")) + data = await call_a2a(backend, {**request_payload, "payload": payload}, self.timeout_seconds) + if key[2]: + self._a2a_sessions[key] = (time.monotonic(), data.get("metadata", {})) + self._a2a_sessions.move_to_end(key) + while len(self._a2a_sessions) > 10000: + self._a2a_sessions.popitem(last=False) + return BackendCallResult( + backend_id=backend.backend_id, backend_url=backend.base_url, + status_code=200, response=data, route_decision=route_decision, + elapsed_ms=int((time.monotonic() - started) * 1000), + ) path = backend.sse_message_path if use_sse else backend.message_path url = f"{backend.base_url}{path}" payload = dict(request_payload) @@ -45,8 +70,16 @@ class BackendClient: ) async def health(self, backend: BackendDefinition) -> dict[str, Any]: - url = f"{backend.base_url}{backend.health_path}" - async with httpx.AsyncClient(timeout=10.0) as client: + path = backend.a2a_agent_card_path if backend.protocol == "a2a" else backend.health_path + url = f"{backend.base_url}{path}" + headers = {} + if backend.protocol == "a2a": + for header, env_name in backend.a2a_headers_env.items(): + value = os.environ.get(env_name) + if not value: + return {"backend_id": backend.backend_id, "ok": False, "error": f"Missing credential environment variable: {env_name}"} + headers[header] = value + async with httpx.AsyncClient(timeout=10.0, headers=headers) as client: try: resp = await client.get(url) return {"backend_id": backend.backend_id, "status_code": resp.status_code, "ok": resp.is_success, "body": self._safe_json(resp)} diff --git a/libs/agent_framework/src/agent_framework/global_supervisor/models.py b/libs/agent_framework/src/agent_framework/global_supervisor/models.py index 99650d36..5b5c676b 100644 --- a/libs/agent_framework/src/agent_framework/global_supervisor/models.py +++ b/libs/agent_framework/src/agent_framework/global_supervisor/models.py @@ -21,6 +21,12 @@ class BackendDefinition(BaseModel): examples: list[str] = Field(default_factory=list) priority: int = 100 enabled: bool = True + protocol: Literal["native", "a2a"] = "native" + a2a_version: Literal["1.0", "0.3"] = "1.0" + a2a_agent_card_path: str = "/.well-known/agent-card.json" + a2a_rpc_path: str = "/a2a" + a2a_headers_env: dict[str, str] = Field(default_factory=dict) + a2a_poll_interval_seconds: float = Field(default=0.5, gt=0) health_path: str = "/health" message_path: str = "/gateway/message" sse_message_path: str = "/gateway/message/sse" diff --git a/templates/agent_template_backend/.env.a2a.example b/templates/agent_template_backend/.env.a2a.example new file mode 100644 index 00000000..e0161dc5 --- /dev/null +++ b/templates/agent_template_backend/.env.a2a.example @@ -0,0 +1,9 @@ +# Complemento opcional ao .env do agente. Ajuste para o endereço público real. +ENABLE_A2A=true +A2A_TASK_MODE=true +A2A_PUBLIC_URL=http://localhost:8000 +A2A_TASK_TIMEOUT_SECONDS=240 +# AGENT_AUTH_* configura o middleware já fornecido pelo framework. +# Exemplo de declaração no Agent Card (não habilita autenticação): +# A2A_SECURITY_SCHEMES={"gatewayBearer":{"httpAuthSecurityScheme":{"scheme":"bearer"}}} +# A2A_SECURITY=[{"gatewayBearer":[]}] diff --git a/templates/agent_template_backend/Dockerfile.a2a b/templates/agent_template_backend/Dockerfile.a2a new file mode 100644 index 00000000..acdc8158 --- /dev/null +++ b/templates/agent_template_backend/Dockerfile.a2a @@ -0,0 +1,8 @@ +FROM python:3.12-slim +WORKDIR /app +COPY libs/agent_framework /agent_framework +COPY templates/agent_template_backend /app +RUN pip install --no-cache-dir -e '/agent_framework[a2a]' -r requirements.txt +ENV ENABLE_A2A=true +ENV A2A_TASK_MODE=true +CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"] diff --git a/templates/agent_template_backend/README.md b/templates/agent_template_backend/README.md index dcfc1558..cef040af 100644 --- a/templates/agent_template_backend/README.md +++ b/templates/agent_template_backend/README.md @@ -4226,3 +4226,10 @@ Este template inclui o arquivo opcional `config/tool_policies.yaml`. Use `operat Além da execução direta de MCP tools, uma operação transacional pode usar um workflow LangGraph determinístico após `clarification` e confirmação explícita. Configure `execution.mode: workflow` em `config/tool_policies.yaml`, mantenha as definições versionadas em `workflows/` e implemente as actions do domínio no projeto do agente. O runtime genérico está em `agent_framework.workflows`. Consulte `libs/agent_framework/docs/TRANSACTIONAL_WORKFLOWS_PT.md` e o exemplo `workflows/devolucao_pedido.v1.yaml`. + +## Agente A2A assíncrono + +O template também pode expor `/a2a`, com Tasks, consulta de status e streaming +SSE, usando `ENABLE_A2A=true` e `A2A_TASK_MODE=true`. Consulte +[A2A assíncrono](../../docs/A2A_ASSINCRONO.md) para ativação e integração ao Gateway. +Os endpoints nativos permanecem disponíveis. diff --git a/templates/agent_template_backend/app/main.py b/templates/agent_template_backend/app/main.py index eab131ec..63f0fe75 100644 --- a/templates/agent_template_backend/app/main.py +++ b/templates/agent_template_backend/app/main.py @@ -168,7 +168,14 @@ def _resolve_identity(req: GatewayRequest, msg) -> tuple[AgentIdentity, dict, Bu return identity, context, business_context, missing_identity_keys -async def _process_gateway_message(req: GatewayRequest, emit_sse: bool = False) -> dict: +async def _emit_gateway_event(session_id, event, payload, emit_sse, progress): + if emit_sse: + await sse_hub.emit(session_id, event, payload) + if progress is not None: + await progress(event, payload) + + +async def _process_gateway_message(req: GatewayRequest, emit_sse: bool = False, progress=None) -> dict: try: msg = await gateway.normalize(req.channel, req.payload) except ValueError as exc: @@ -191,7 +198,7 @@ async def _process_gateway_message(req: GatewayRequest, emit_sse: bool = False) stream = sse_hub.stream_for(agent_session_id) async with stream.lock: - await sse_hub.emit(agent_session_id, "flow.start", {"session_id": agent_session_id, "message_id": message_id, "agent_id": identity.agent_id}) if emit_sse else None + await _emit_gateway_event(agent_session_id, "flow.start", {"session_id": agent_session_id, "message_id": message_id, "agent_id": identity.agent_id}, emit_sse, progress) session = await sessions.get(agent_session_id) if not session: @@ -296,11 +303,11 @@ async def _process_gateway_message(req: GatewayRequest, emit_sse: bool = False) ) rendered = await gateway.render(response) await telemetry.event("gateway.message.replayed", {"session_id": agent_session_id, "reason": interruption.reason}) - await sse_hub.emit(agent_session_id, "message.responded", rendered) if emit_sse else None + await _emit_gateway_event(agent_session_id, "message.responded", rendered, emit_sse, progress) return rendered effective_text = interruption.text - await sse_hub.emit(agent_session_id, "session.upserted", {"session_id": agent_session_id, "business_context": business_context.model_dump()}) if emit_sse else None + await _emit_gateway_event(agent_session_id, "session.upserted", {"session_id": agent_session_id, "business_context": business_context.model_dump()}, emit_sse, progress) await memory.append( agent_session_id, @@ -317,7 +324,7 @@ async def _process_gateway_message(req: GatewayRequest, emit_sse: bool = False) }, ), ) - await sse_hub.emit(agent_session_id, "message.received", {"session_id": agent_session_id, "role": "user"}) if emit_sse else None + await _emit_gateway_event(agent_session_id, "message.received", {"session_id": agent_session_id, "role": "user"}, emit_sse, progress) history = [m.model_dump(mode="json") for m in await memory.list(agent_session_id)] cms_input = { @@ -362,7 +369,7 @@ async def _process_gateway_message(req: GatewayRequest, emit_sse: bool = False) _root_span=True, ) as root_span: await telemetry.event("gateway.message.received", trace_context) - await sse_hub.emit(agent_session_id, "workflow.started", trace_context) if emit_sse else None + await _emit_gateway_event(agent_session_id, "workflow.started", trace_context, emit_sse, progress) result = await workflow.ainvoke( { "tenant_id": identity.tenant_id, @@ -396,7 +403,7 @@ async def _process_gateway_message(req: GatewayRequest, emit_sse: bool = False) ) await checkpoints.put(agent_session_id, {"state": result, "message_id": message_id}) - await sse_hub.emit(agent_session_id, "workflow.completed", {"session_id": agent_session_id, "route": result.get("route"), "intent": result.get("intent")}) if emit_sse else None + await _emit_gateway_event(agent_session_id, "workflow.completed", {"session_id": agent_session_id, "route": result.get("route"), "intent": result.get("intent")}, emit_sse, progress) answer = result.get("final_answer") or result.get("answer") or "" await memory.append( @@ -480,8 +487,8 @@ async def _process_gateway_message(req: GatewayRequest, emit_sse: bool = False) ) rendered = await gateway.render(response) root_span.set_output(rendered) - await sse_hub.emit(agent_session_id, "message.responded", rendered) if emit_sse else None - await sse_hub.emit(agent_session_id, "flow.end", {"session_id": agent_session_id, "message_id": message_id}) if emit_sse else None + await _emit_gateway_event(agent_session_id, "message.responded", rendered, emit_sse, progress) + await _emit_gateway_event(agent_session_id, "flow.end", {"session_id": agent_session_id, "message_id": message_id}, emit_sse, progress) return rendered @@ -649,3 +656,17 @@ async def get_session_checkpoint(session_id: str): @app.on_event("shutdown") async def shutdown(): telemetry.shutdown() + + +# Optional direct A2A endpoint for agents built with the framework template. +if settings.ENABLE_A2A: + from agent_framework.a2a.progress import report_progress + from agent_framework.a2a.server import install_backend_a2a + from agent_framework.security import install_authentication + + install_authentication(app, prefix="AGENT_AUTH") + + async def _a2a_backend_handler(data: dict): + return await _process_gateway_message(GatewayRequest(**data), progress=report_progress) + + install_backend_a2a(app, _a2a_backend_handler, agent_profiles, settings) diff --git a/tests/test_a2a_async_tasks.py b/tests/test_a2a_async_tasks.py new file mode 100644 index 00000000..8aaa6a94 --- /dev/null +++ b/tests/test_a2a_async_tasks.py @@ -0,0 +1,258 @@ +"""Real SDK/FastAPI tests: no mock task manager or custom JSON-RPC dispatcher.""" +import asyncio +import json +import threading +import time +from types import SimpleNamespace + +import pytest +from fastapi import FastAPI +from fastapi.testclient import TestClient +pytest.importorskip("a2a", reason="Install agent-framework[a2a] to run A2A tests") + +from agent_framework.a2a.server import install_a2a, install_backend_a2a +from agent_framework.a2a.progress import report_progress +from agent_framework.global_supervisor.models import BackendDefinition + + +HEADERS = {'A2A-Version': '1.0'} + + +def send(client, method, params, legacy=False): + return client.post('/a2a', headers={'A2A-Version': '0.3' if legacy else '1.0'}, json={ + 'jsonrpc': '2.0', 'id': 'request', 'method': method, 'params': params, + }) + + +def message_params(legacy=False, immediate=True, task_id=None): + message = {'messageId': 'm-' + str(time.monotonic_ns()), 'contextId': 'session', + 'role': 'user' if legacy else 'ROLE_USER', 'parts': [{'text': 'Olá'}]} + if task_id: + message['taskId'] = task_id + if legacy: + message['kind'] = 'message' + message['parts'][0]['kind'] = 'text' + return {'message': message, + 'configuration': {'blocking': not immediate} if legacy else {'returnImmediately': immediate}} + + +def task_result(response, legacy=False): + data = response.json() + assert 'error' not in data, data + return data['result'] if legacy else data['result']['task'] + + +def get_task(client, task_id, legacy=False): + data = send(client, 'tasks/get' if legacy else 'GetTask', {'id': task_id}, legacy).json() + assert 'error' not in data, data + return data['result'] + + +def wait_state(client, task_id, expected, legacy=False): + deadline = time.monotonic() + 3 + while time.monotonic() < deadline: + task = get_task(client, task_id, legacy) + if task['status']['state'] == expected: + return task + time.sleep(.005) + pytest.fail(f'Task did not reach {expected}: {task}') + + +def application(handler, **kwargs): + app = FastAPI() + registry = SimpleNamespace(list=lambda: [BackendDefinition(backend_id='native', url='http://native')]) + install_a2a(app, handler, registry, 'http://testserver', **kwargs) + return app + + +@pytest.mark.parametrize('legacy', [False, True]) +def test_submit_returns_before_processing_and_poll_after_disconnect(legacy): + release = threading.Event() + async def handler(request): + while not release.is_set(): + await asyncio.sleep(.005) + return {'text': 'done', 'session_id': request['session_id']} + with TestClient(application(handler)) as client: + try: + first = task_result(send(client, 'message/send' if legacy else 'SendMessage', message_params(legacy), legacy), legacy) + assert not release.is_set() + assert first['status']['state'] in ({'submitted', 'working'} if legacy else {'TASK_STATE_SUBMITTED', 'TASK_STATE_WORKING'}) + assert get_task(client, first['id'], legacy)['contextId'] == 'session' + release.set() + task = wait_state(client, first['id'], 'completed' if legacy else 'TASK_STATE_COMPLETED', legacy) + assert task['artifacts'][0]['parts'][0]['text'] == 'done' + finally: + release.set() + + +@pytest.mark.parametrize('legacy', [False, True]) +def test_stream_contains_task_progress_artifact_and_terminal_state(legacy): + async def handler(request): + await report_progress('workflow.started', {'route': 'orders'}) + await asyncio.sleep(.01) + await report_progress('workflow.completed', {'route': 'orders'}) + return {'text': 'done'} + with TestClient(application(handler)) as client: + assert client.get('/.well-known/agent-card.json').json()['capabilities']['streaming'] + response = send(client, 'message/stream' if legacy else 'SendStreamingMessage', message_params(legacy, immediate=False), legacy) + assert 'text/event-stream' in response.headers['content-type'], response.text + events = [json.loads(line[5:].strip()) for line in response.text.splitlines() if line.startswith('data:')] + assert events and all('error' not in e for e in events), events + text = json.dumps(events) + assert 'workflow.started' in text and 'workflow.completed' in text + assert 'done' in text + assert ('completed' if legacy else 'TASK_STATE_COMPLETED') in text + first = events[0]['result'] + assert ('id' in first if legacy else 'task' in first), events + + +def test_cancel_stops_local_execution(): + started, stopped = threading.Event(), threading.Event() + async def handler(request): + started.set() + try: + await asyncio.sleep(30) + finally: + stopped.set() + with TestClient(application(handler)) as client: + first = task_result(send(client, 'SendMessage', message_params())) + assert started.wait(1) + result = send(client, 'CancelTask', {'id': first['id']}).json() + assert 'error' not in result, result + assert result['result']['status']['state'] == 'TASK_STATE_CANCELED' + assert stopped.wait(1) + assert get_task(client, first['id'])['status']['state'] == 'TASK_STATE_CANCELED' + + +@pytest.mark.parametrize('failure', ['exception', 'timeout']) +def test_failure_is_persisted(failure): + async def handler(request): + if failure == 'exception': + raise RuntimeError('internal-sensitive-detail') + await asyncio.sleep(1) + with TestClient(application(handler, timeout_seconds=.02)) as client: + first = task_result(send(client, 'SendMessage', message_params())) + task = wait_state(client, first['id'], 'TASK_STATE_FAILED') + assert 'internal-sensitive-detail' not in json.dumps(task) + assert task['status']['message']['parts'][0]['text'] == ('Processing timed out' if failure == 'timeout' else 'Agent processing failed') + + +def test_input_required_resumes_same_task(): + turns = [] + async def handler(request): + turns.append(request) + return {'text': 'Confirmar?' if len(turns) == 1 else 'Concluído', + 'metadata': {'a2a_task_state': 'input-required'} if len(turns) == 1 else {}} + with TestClient(application(handler)) as client: + first = task_result(send(client, 'SendMessage', message_params(immediate=False))) + assert first['status']['state'] == 'TASK_STATE_INPUT_REQUIRED' + second = task_result(send(client, 'SendMessage', message_params(immediate=False, task_id=first['id']))) + assert second['id'] == first['id'] + assert second['status']['state'] == 'TASK_STATE_COMPLETED' + assert len(turns) == 2 + # A completed task cannot execute a third business operation. + response = send(client, 'SendMessage', message_params(task_id=first['id'])).json() + assert 'error' in response + assert len(turns) == 2 + + +def test_backend_installer_exposes_agent_skills(): + seen = [] + async def handler(request): + seen.append(request) + return {'text': 'native agent', 'session_id': request['session_id']} + profiles = SimpleNamespace(list_profiles=lambda: [SimpleNamespace(agent_id='orders', name='Orders', description='Order status')]) + settings = SimpleNamespace(A2A_PUBLIC_URL='https://backend.example', A2A_SECURITY_SCHEMES={}, A2A_SECURITY=[], + A2A_TASK_MODE=True, A2A_TASK_TIMEOUT_SECONDS=10, APP_NAME='Framework backend') + app = FastAPI() + install_backend_a2a(app, handler, profiles, settings) + with TestClient(app) as client: + card = client.get('/.well-known/agent-card.json').json() + assert card['skills'][0]['id'] == 'orders' + assert card['supportedInterfaces'][0]['url'] == 'https://backend.example/a2a' + params = message_params(immediate=False) + params['message']['metadata'] = {'agent_id': 'orders', 'tenant_id': 't'} + final = task_result(send(client, 'SendMessage', params)) + assert final['status']['state'] == 'TASK_STATE_COMPLETED' + assert seen[0]['agent_id'] == 'orders' and seen[0]['tenant_id'] == 't' + + +def test_tasks_are_scoped_to_authenticated_principal(): + from agent_framework.security.middleware import AuthenticationMiddleware + from agent_framework.security.authentication import AuthenticationResult, AuthenticatedPrincipal + + class TestProvider: + async def authenticate(self, request): + # Test-only identity provider, equivalent to verified middleware output. + return AuthenticationResult(True, AuthenticatedPrincipal(request.headers['x-test-principal'], 'test')) + + async def handler(request): + return {'text': 'private result'} + app = application(handler) + app.add_middleware(AuthenticationMiddleware, provider=TestProvider()) + with TestClient(app) as client: + client.headers['x-test-principal'] = 'alice' + first = task_result(send(client, 'SendMessage', message_params(immediate=False))) + assert get_task(client, first['id'])['artifacts'][0]['parts'][0]['text'] == 'private result' + client.headers['x-test-principal'] = 'bob' + for method in ['GetTask', 'CancelTask']: + result = send(client, method, {'id': first['id']}).json() + assert 'error' in result + assert 'private result' not in json.dumps(result) + listing = send(client, 'ListTasks', {}).json() + assert not listing['result'].get('tasks'), listing + + +@pytest.mark.parametrize('legacy', [False, True]) +def test_gateway_to_framework_a2a_backend(monkeypatch, legacy): + import httpx + from agent_framework.global_supervisor.client import BackendClient + from agent_framework.global_supervisor.models import GlobalRouteDecision + backend_release = threading.Event() + async def backend_handler(request): + while not backend_release.is_set(): + await asyncio.sleep(.005) + return {'text': 'backend complete', 'session_id': request['session_id']} + backend_app = application(backend_handler) + real_client = httpx.AsyncClient + monkeypatch.setattr(httpx, 'AsyncClient', lambda **kw: real_client(transport=httpx.ASGITransport(app=backend_app), **kw)) + backend_client = BackendClient(timeout_seconds=2) + definition = BackendDefinition(backend_id='x', url='http://backend', protocol='a2a', + a2a_version='0.3' if legacy else '1.0', a2a_poll_interval_seconds=.005) + async def gateway_handler(request): + result = await backend_client.call_message(definition, request, GlobalRouteDecision(backend_id='x')) + return result.response + with TestClient(application(gateway_handler)) as client: + try: + first = task_result(send(client, 'SendMessage', message_params())) + assert first['status']['state'] in {'TASK_STATE_SUBMITTED', 'TASK_STATE_WORKING'} + assert not backend_release.is_set() + backend_release.set() + final = wait_state(client, first['id'], 'TASK_STATE_COMPLETED') + assert final['artifacts'][0]['parts'][0]['text'] == 'backend complete' + finally: + backend_release.set() + + +def test_template_progress_helper_preserves_native_event_contract(): + # Exercise the actual lightweight helper without booting OCI/LLM dependencies. + import ast + from pathlib import Path + source = Path(__file__).resolve().parents[1] / 'templates/agent_template_backend/app/main.py' + module = ast.parse(source.read_text()) + helper = next(n for n in module.body if isinstance(n, ast.AsyncFunctionDef) and n.name == '_emit_gateway_event') + native, a2a = [], [] + async def native_emit(*args): + native.append(args) + async def progress(*args): + a2a.append(args) + namespace = {'sse_hub': SimpleNamespace(emit=native_emit)} + exec(compile(ast.Module(body=[helper], type_ignores=[]), str(source), 'exec'), namespace) + emit = namespace['_emit_gateway_event'] + async def scenario(): + await emit('session', 'flow.start', {'x': 1}, True, None) + await emit('session', 'workflow.started', {'x': 2}, False, progress) + await emit('session', 'message.responded', {'x': 3}, False, None) + asyncio.run(scenario()) + assert native == [('session', 'flow.start', {'x': 1})] + assert a2a == [('workflow.started', {'x': 2})] diff --git a/tests/test_a2a_http_streaming.py b/tests/test_a2a_http_streaming.py new file mode 100644 index 00000000..ad11de1c --- /dev/null +++ b/tests/test_a2a_http_streaming.py @@ -0,0 +1,81 @@ +"""Verify SSE flush and disconnect/reconnect over a real local HTTP connection.""" +import asyncio +import json +import socket +import threading +import time +from types import SimpleNamespace + +import httpx +import pytest +import uvicorn +from fastapi import FastAPI +pytest.importorskip("a2a", reason="Install agent-framework[a2a] to run A2A tests") + +from agent_framework.a2a.server import install_a2a +from agent_framework.global_supervisor.models import BackendDefinition + + +@pytest.mark.parametrize('legacy', [False, True]) +def test_sse_flush_disconnect_and_resubscribe(legacy): + release = threading.Event() + finished = threading.Event() + async def handler(request): + while not release.is_set(): + await asyncio.sleep(.005) + finished.set() + return {'text': 'stream complete'} + app = FastAPI() + registry = SimpleNamespace(list=lambda: [BackendDefinition(backend_id='x', url='http://native')]) + install_a2a(app, handler, registry, 'http://testserver') + sock = socket.socket() + sock.bind(('127.0.0.1', 0)) + server = uvicorn.Server(uvicorn.Config(app, log_level='error', lifespan='off')) + thread = threading.Thread(target=lambda: server.run(sockets=[sock]), daemon=True) + thread.start() + try: + deadline = time.monotonic() + 3 + while not server.started and time.monotonic() < deadline: + time.sleep(.005) + assert server.started + base_url = f'http://127.0.0.1:{sock.getsockname()[1]}' + with httpx.Client(base_url=base_url, headers={'A2A-Version': '0.3' if legacy else '1.0'}, timeout=2, trust_env=False) as client: + message = {'messageId': 'm', 'role': 'user' if legacy else 'ROLE_USER', + 'parts': [{'kind': 'text', 'text': 'run'}] if legacy else [{'text': 'run'}]} + if legacy: message['kind'] = 'message' + def envelope(method, params): + return {'jsonrpc': '2.0', 'id': 1, 'method': method, 'params': params} + with client.stream('POST', '/a2a', json=envelope('message/stream' if legacy else 'SendStreamingMessage', {'message': message})) as response: + assert response.status_code == 200 + for line in response.iter_lines(): + if line.startswith('data:'): + event = json.loads(line[5:].strip()) + first = event['result'] if legacy else event['result']['task'] + task_id = first['id'] + break + else: + pytest.fail('SSE did not deliver an initial Task') + assert not release.is_set() and not finished.is_set() + # Closing the first SSE connection must not cancel the agent task. + current = client.post('/a2a', json=envelope('tasks/get' if legacy else 'GetTask', {'id': task_id})).json() + assert current['result']['status']['state'] in ({'submitted', 'working'} if legacy else {'TASK_STATE_SUBMITTED', 'TASK_STATE_WORKING'}) + with client.stream('POST', '/a2a', json=envelope('tasks/resubscribe' if legacy else 'SubscribeToTask', {'id': task_id})) as response: + events = [] + for line in response.iter_lines(): + if not line.startswith('data:'): + continue + event = json.loads(line[5:].strip()) + assert 'error' not in event, event + events.append(event) + if len(events) == 1: + release.set() + encoded = json.dumps(events) + assert 'stream complete' in encoded + assert ('completed' if legacy else 'TASK_STATE_COMPLETED') in encoded + assert finished.is_set() + finally: + release.set() + server.should_exit = True + thread.join(timeout=3) + sock.close() + assert not thread.is_alive() diff --git a/tests/test_a2a_interoperability.py b/tests/test_a2a_interoperability.py new file mode 100644 index 00000000..d576c0d2 --- /dev/null +++ b/tests/test_a2a_interoperability.py @@ -0,0 +1,188 @@ +import asyncio +import json +from types import SimpleNamespace + +import httpx +import pytest +from fastapi import FastAPI +from fastapi.testclient import TestClient +pytest.importorskip("a2a", reason="Install agent-framework[a2a] to run A2A tests") + +from agent_framework.a2a.server import install_a2a +from agent_framework.global_supervisor.models import BackendDefinition, GlobalRouteDecision +from agent_framework.global_supervisor.client import BackendClient + + +@pytest.fixture +def facade(): + seen = [] + async def handler(request): + seen.append(request) + return {'channel': 'web', 'session_id': request['session_id'], 'text': 'Olá', 'metadata': {}} + app = FastAPI() + registry = SimpleNamespace(list=lambda: [BackendDefinition(backend_id='native', url='http://native')]) + install_a2a(app, handler, registry, 'https://agents.example', task_mode=False) + with TestClient(app) as client: + yield client, seen + + +@pytest.mark.parametrize('legacy', [False, True]) +def test_discovery_and_send(facade, legacy): + client, seen = facade + card = client.get('/.well-known/agent-card.json').json() + assert card['supportedInterfaces'][0]['protocolVersion'] == '1.0' + message = {'messageId': 'm', 'role': 'ROLE_USER', 'contextId': 'conversation', + 'parts': [{'text': 'Oi'}, {'data': {'invoice_id': 42, 'session_id': 'override'}}]} + if legacy: + message.update(kind='message', role='user') + for part in message['parts']: part['kind'] = 'text' if 'text' in part else 'data' + response = client.post('/a2a', headers={'A2A-Version': '0.3' if legacy else '1.0'}, json={ + 'jsonrpc': '2.0', 'id': 'x', 'method': 'message/send' if legacy else 'SendMessage', + 'params': {'message': message}, + }).json() + reply = response['result'] if legacy else response['result']['message'] + assert reply['parts'][0]['text'] == 'Olá', response + assert reply['contextId'] == 'conversation' + assert seen[0]['payload']['invoice_id'] == 42 + assert seen[0]['payload']['session_id'] == 'conversation' + + +def test_errors(facade): + client, _ = facade + for method, params in [('Unknown', {}), ('GetTask', {'id': 'missing'})]: + response = client.post('/a2a', headers={'A2A-Version': '1.0'}, json={ + 'jsonrpc': '2.0', 'id': 1, 'method': method, 'params': params, + }).json() + assert 'error' in response + + +def mock_http(monkeypatch, handler): + real_client = httpx.AsyncClient + monkeypatch.setattr(httpx, 'AsyncClient', lambda **kw: real_client(transport=httpx.MockTransport(handler), **kw)) + + +@pytest.mark.parametrize('protocol', ['native', '1.0', '0.3']) +def test_backend_adapter(monkeypatch, protocol): + seen = [] + def transport(request): + body = json.loads(request.content) + seen.append((request, body)) + if protocol == 'native': return httpx.Response(200, json={'text': 'OK', 'session_id': 's'}) + reply = {'messageId': 'r', 'role': 'ROLE_AGENT', 'contextId': 'remote', 'parts': [{'text': 'OK'}]} + if protocol == '0.3': + reply.update(kind='message', role='agent'); reply['parts'][0]['kind'] = 'text' + return httpx.Response(200, json={'jsonrpc': '2.0', 'id': body['id'], 'result': reply if protocol == '0.3' else {'message': reply}}) + mock_http(monkeypatch, transport) + backend = BackendDefinition(backend_id='x', url='http://backend', protocol='native' if protocol == 'native' else 'a2a', a2a_version='0.3' if protocol == '0.3' else '1.0') + response = asyncio.run(BackendClient().call_message(backend, {'payload': {'session_id': 's', 'message': 'Oi'}}, GlobalRouteDecision(backend_id='x'))) + assert response.response['text'] == 'OK' + assert response.response['session_id'] == 's' + if protocol == 'native': + assert str(seen[0][0].url) == 'http://backend/gateway/message' + assert seen[0][1]['payload']['selected_backend'] == 'x' + else: + assert seen[0][1]['method'] == ('message/send' if protocol == '0.3' else 'SendMessage') + + +def test_task_polling(monkeypatch): + methods = [] + def transport(request): + body = json.loads(request.content); methods.append(body['method']) + task = {'id': 'task', 'contextId': 'remote', 'status': {'state': 'TASK_STATE_WORKING' if len(methods) == 1 else 'TASK_STATE_COMPLETED'}} + if len(methods) == 2: task['artifacts'] = [{'artifactId': 'a', 'parts': [{'text': 'done'}]}] + return httpx.Response(200, json={'jsonrpc': '2.0', 'id': body['id'], 'result': {'task': task} if len(methods) == 1 else task}) + mock_http(monkeypatch, transport) + backend = BackendDefinition(backend_id='x', url='http://backend', protocol='a2a', a2a_poll_interval_seconds=.001) + response = asyncio.run(BackendClient().call_message(backend, {'payload': {'message': 'Oi'}}, GlobalRouteDecision(backend_id='x'))) + assert response.response['text'] == 'done' + assert methods == ['SendMessage', 'GetTask'] + + +def test_rpc_failure(monkeypatch): + def transport(request): + body = json.loads(request.content) + return httpx.Response(200, json={'jsonrpc': '2.0', 'id': body['id'], 'error': {'code': -32603, 'message': 'private'}}) + mock_http(monkeypatch, transport) + backend = BackendDefinition(backend_id='x', url='http://backend', protocol='a2a') + with pytest.raises(RuntimeError, match='-32603'): + asyncio.run(BackendClient().call_message(backend, {'payload': {}}, GlobalRouteDecision(backend_id='x'))) + + +def test_continuation_and_tenant_isolation(monkeypatch): + seen = [] + def transport(request): + body = json.loads(request.content); seen.append(body['params']['message']) + return httpx.Response(200, json={'jsonrpc': '2.0', 'id': body['id'], 'result': {'task': { + 'id': 'remote-task', 'contextId': 'remote-context', 'status': {'state': 'TASK_STATE_INPUT_REQUIRED', + 'message': {'messageId': 'a', 'role': 'ROLE_AGENT', 'parts': [{'text': 'Confirmar?'}]}} + }}}) + mock_http(monkeypatch, transport) + backend = BackendDefinition(backend_id='x', url='http://backend', protocol='a2a') + async def scenario(): + client = BackendClient() + for tenant in ['first', 'first', 'second']: + await client.call_message(backend, {'tenant_id': tenant, 'payload': {'session_id': 'same', 'message': 'sim'}}, GlobalRouteDecision(backend_id='x')) + asyncio.run(scenario()) + assert seen[1]['taskId'] == 'remote-task' + assert seen[1]['contextId'] == 'remote-context' + assert 'taskId' not in seen[2] + assert seen[0]['contextId'] != seen[2]['contextId'] + + +def test_credentials_and_deadline(monkeypatch): + monkeypatch.delenv('MISSING_A2A_TOKEN', raising=False) + backend = BackendDefinition(backend_id='x', url='http://backend', protocol='a2a', a2a_headers_env={'Authorization': 'MISSING_A2A_TOKEN'}) + with pytest.raises(ValueError, match='MISSING_A2A_TOKEN'): + asyncio.run(BackendClient().call_message(backend, {'payload': {}}, GlobalRouteDecision(backend_id='x'))) + async def transport(request): + await asyncio.sleep(1) + return httpx.Response(200) + mock_http(monkeypatch, transport) + backend.a2a_headers_env = {} + with pytest.raises(TimeoutError): + asyncio.run(BackendClient(timeout_seconds=.01).call_message(backend, {'payload': {}}, GlobalRouteDecision(backend_id='x'))) + + +def test_native_does_not_import_sdk(): + import subprocess + import sys + code = ''' +import sys +class BlockA2A: + def find_spec(self, fullname, *args): + if fullname == 'a2a' or fullname.startswith('a2a.'): + raise ImportError('A2A unavailable') +sys.meta_path.insert(0, BlockA2A()) +from agent_framework.global_supervisor.client import BackendClient +from agent_framework.global_supervisor.models import BackendDefinition +assert BackendDefinition(backend_id='x', url='http://x').protocol == 'native' +BackendClient() +''' + subprocess.run([sys.executable, '-c', code], check=True) + + +def test_health_uses_card_and_credentials(monkeypatch): + seen = [] + def transport(request): + seen.append(request) + return httpx.Response(200, json={'name': 'external'}) + mock_http(monkeypatch, transport) + monkeypatch.setenv('A2A_TEST_AUTH', 'Bearer test') + backend = BackendDefinition(backend_id='x', url='http://backend', protocol='a2a', a2a_headers_env={'Authorization':'A2A_TEST_AUTH'}) + result = asyncio.run(BackendClient().health(backend)) + assert result['ok'] + assert seen[0].url.path == '/.well-known/agent-card.json' + assert seen[0].headers['Authorization'] == 'Bearer test' + + +def test_a2a_backend_selection_is_forwarded_to_gateway(facade): + client, seen = facade + response = client.post('/a2a', headers={'A2A-Version': '1.0'}, json={ + 'jsonrpc': '2.0', 'id': 'select', 'method': 'SendMessage', 'params': { + 'message': {'messageId': 'm', 'role': 'ROLE_USER', 'parts': [{'text': 'Olá'}], + 'metadata': {'backend_id': 'billing', 'agent_id': 'billing_agent'}}, + }, + }).json() + assert 'result' in response, response + assert seen[0]['backend_id'] == 'billing' + assert seen[0]['agent_id'] == 'billing_agent'