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
+
+
+
+
+
+
+
+ Pronto para conectar
+
+
+
+
+
+ ✳CONVERSAS QUE VIRAM AÇÕESO próximo passo
começa com uma mensagem.
Conecte um agente ou o Gateway do framework.
Acompanhe cada tarefa até a resposta.
+
+
+ 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'