A2A Protocol (Agent-to-Agent)

OverviewCopy Anchor Link

The @cogitator-ai/a2a package implements Google's A2A Protocol v0.3 — the open standard for agent-to-agent communication. It enables Cogitator agents to interoperate with agents from any framework (LangChain, CrewAI, etc.) via a standardized JSON-RPC protocol.

• A2AServer -- expose any Cogitator agent as an A2A-compliant service with auto-generated Agent Cards
• A2AClient -- connect to any remote A2A agent, with discovery, messaging, and streaming
• asTool() Bridge -- wrap remote A2A agents as local Cogitator tools for seamless delegation
• Framework Adapters -- Express, Hono, Fastify, Koa, Next.js

pnpm add @cogitator-ai/a2a

A2AServerCopy Anchor Link

Exposing AgentsCopy Anchor Link

Create an A2A server to expose your agents to any A2A-compatible client:

import { Cogitator, Agent } from '@cogitator-ai/core';
import { A2AServer } from '@cogitator-ai/a2a';
import { a2aExpress } from '@cogitator-ai/a2a/express';
import express from 'express';

const cogitator = new Cogitator();

const researcher = new Agent({
  name: 'researcher',
  description: 'Research agent that finds information',
  model: 'openai/gpt-6.1-sol',
  instructions: 'You are a research assistant.',
});

const a2aServer = new A2AServer({
  agents: { researcher },
  cogitator,
  cardUrl: 'https://my-server.com',
});

const app = express();
app.use(a2aExpress(a2aServer));
app.listen(3000);

This exposes:

• GET /.well-known/agent.json -- Agent Card discovery
• POST /a2a -- JSON-RPC endpoint for messaging, streaming, and task management

basePath (default /a2a, must start with /) is the JSON-RPC path the adapters serve, relative to where they are mounted, and the card URL when cardUrl is unset; the server exposes it as a2aServer.basePath. For Next.js, put the POST route file at that path. cardUrl is the url the cards advertise.

const a2aServer = new A2AServer({ agents: { researcher }, cogitator, basePath: '/rpc' });

app.use(a2aExpress(a2aServer)); // POST /rpc
const client = new A2AClient('http://localhost:3000', { rpcPath: a2aServer.basePath });

Agent CardsCopy Anchor Link

Agent Cards are auto-generated from your Agent configuration. Each tool becomes a skill in the card:

const card = a2aServer.getAgentCard('researcher');

{
  "name": "researcher",
  "description": "Research agent that finds information",
  "url": "https://my-server.com",
  "version": "0.3",
  "capabilities": {
    "streaming": true,
    "pushNotifications": false,
    "extendedAgentCard": false
  },
  "skills": [
    {
      "id": "web_search",
      "name": "web_search",
      "description": "Search the web for information",
      "inputModes": ["text/plain", "application/json"],
      "outputModes": ["text/plain", "application/json"]
    }
  ],
  "defaultInputModes": ["text/plain"],
  "defaultOutputModes": ["text/plain", "application/json"]
}

capabilities.pushNotifications is true when you pass a pushNotificationStore, and extendedAgentCard when you pass an extendedCardGenerator. With auth, the card also lists securitySchemes and security. getAgentCards() returns every card, signed when cardSigning is set.

Multiple AgentsCopy Anchor Link

Register multiple agents on a single server (at least one is required):

const a2aServer = new A2AServer({
  agents: {
    researcher: researchAgent,
    writer: writerAgent,
    analyst: analystAgent,
  },
  cogitator,
});

/.well-known/agent.json then returns an array of cards. message/send, message/stream and agent/extendedCard pick the agent from an agentName param and fall back to the first registered agent. To reach another agent from A2AClient, pass its registered name:

const writer = new A2AClient('https://my-server.com', { agentName: 'writer' });

const card = await writer.agentCard();
const task = await writer.sendMessage({
  role: 'user',
  parts: [{ type: 'text', text: 'Draft a summary' }],
});

The client sends agentName with message/send, message/stream and agent/extendedCard, and agentCard() returns the card of that name (-32007 Agent not found when the server has none).

A2AClientCopy Anchor Link

Connecting to a Remote AgentCopy Anchor Link

import { A2AClient } from '@cogitator-ai/a2a';

const client = new A2AClient('https://remote-agent.example.com');

const card = await client.agentCard();
console.log(`Agent: ${card.name}`);
console.log(`Skills: ${card.skills.map((s) => s.name).join(', ')}`);

Sending MessagesCopy Anchor Link

const task = await client.sendMessage({
  role: 'user',
  parts: [{ type: 'text', text: 'Research quantum computing trends' }],
});

console.log(task.status.state);

for (const artifact of task.artifacts) {
  for (const part of artifact.parts) {
    if (part.type === 'text') console.log(part.text);
  }
}

StreamingCopy Anchor Link

for await (const event of client.sendMessageStream({
  role: 'user',
  parts: [{ type: 'text', text: 'Analyze market trends' }],
})) {
  if (event.type === 'status-update') {
    console.log(`Status: ${event.status.state}`);
  } else if (event.type === 'artifact-update') {
    console.log(`Artifact: ${event.artifact.id}`);
  }
}

Artifact events arrive before the final status, and the stream ends on a terminal state or input-required. The client timeout is an idle timeout between events; pass { signal } as the third argument to cancel (the server aborts the agent run when the client disconnects).

Send ConfigurationCopy Anchor Link

const task = await client.sendMessage(
  { role: 'user', parts: [{ type: 'text', text: 'Long running analysis' }] },
  { blocking: false, pushNotificationConfig: { webhookUrl: 'https://my-app.com/hooks/a2a' } }
);

Task ManagementCopy Anchor Link

const task = await client.getTask('task_abc123', 5);

await client.cancelTask('task_abc123');

asTool() BridgeCopy Anchor Link

Wrap any remote A2A agent as a local Cogitator tool:

const client = new A2AClient('https://research-agent.example.com');
const card = await client.agentCard();

const remoteTool = client.asToolFromCard(card, {
  name: 'remote_researcher',
  description: 'Delegate research to remote agent',
});

const orchestrator = new Agent({
  name: 'orchestrator',
  model: 'openai/gpt-6.1-sol',
  instructions: 'Use the remote_researcher for information gathering.',
  tools: [remoteTool],
});

const result = await cogitator.run(orchestrator, {
  input: 'Write a report on AI trends',
});

The orchestrator agent calls the remote researcher via A2A protocol transparently -- it looks like a regular tool call with a single task string parameter. asToolFromCard names and describes the tool from the card unless you override them; client.asTool({ name?, description?, timeout? }) does the same without a card (default name a2a_remote_agent). The tool is marked with the external side effect and forwards the run's abort signal.

The tool result is { output, success, error?, taskId?, state? }; when the remote agent asks a question (state: 'input-required'), output holds the question and taskId lets you continue the task.

Multi-Turn ConversationsCopy Anchor Link

A2A supports multi-turn conversations where follow-up messages continue an existing task. Each task belongs to a contextId that groups related tasks into a conversation.

Client-SideCopy Anchor Link

Use continueTask to send follow-up messages to an existing task:

const client = new A2AClient('https://remote-agent.example.com');

const task = await client.sendMessage({
  role: 'user',
  parts: [{ type: 'text', text: 'Research quantum computing' }],
});

const followUp = await client.continueTask(task.id, 'Now compare it with classical computing');

console.log(task.contextId === followUp.contextId); // true

Server-SideCopy Anchor Link

On the server, multi-turn is handled automatically. When a message includes a taskId, the server appends the new message to the existing task history and re-runs the agent with the task transcript as context, so follow-ups work even without a memory adapter. Runs are threaded by contextId, artifacts accumulate across turns, and a task that is still running cannot be continued concurrently.

Listing TasksCopy Anchor Link

Query stored tasks with filtering and pagination via the tasks/list method:

const client = new A2AClient('https://remote-agent.example.com');

const allTasks = await client.listTasks();

const conversationTasks = await client.listTasks({
  contextId: 'ctx_abc123',
});

const completedTasks = await client.listTasks({
  state: 'completed',
  limit: 10,
  offset: 0,
});

Tasks are returned sorted by timestamp (newest first). The TaskFilter supports:

Token StreamingCopy Anchor Link

SSE streaming now includes TokenStreamEvent alongside status and artifact updates, giving clients real-time character-by-character output:

for await (const event of client.sendMessageStream({
  role: 'user',
  parts: [{ type: 'text', text: 'Write a poem about the ocean' }],
})) {
  switch (event.type) {
    case 'token':
      process.stdout.write(event.token);
      break;
    case 'status-update':
      console.log(`\nStatus: ${event.status.state}`);
      break;
    case 'artifact-update':
      console.log(`\nArtifact: ${event.artifact.id}`);
      break;
  }
}

The A2AStreamEvent union type includes three event types:

The server emits token events via the onToken callback passed to Cogitator.run().

RedisTaskStoreCopy Anchor Link

For production deployments, use RedisTaskStore instead of the default in-memory store:

import { A2AServer, RedisTaskStore } from '@cogitator-ai/a2a';
import { createRedisClient } from '@cogitator-ai/redis';

const redis = await createRedisClient({ url: 'redis://localhost:6379' });

const a2aServer = new A2AServer({
  agents: { researcher },
  cogitator,
  taskStore: new RedisTaskStore({
    client: redis,
    keyPrefix: 'a2a:task:',
    ttl: 86400,
  }),
});

RedisTaskStore accepts any Redis client that implements the RedisClientLike interface: get, set, del, keys, and optionally setex (TTL), scan (listing without blocking KEYS), mget (batched reads) and eval (atomic task updates). The client from @cogitator-ai/redis (createRedisClient, standalone or cluster) and an ioredis instance (new Redis(url)) fit it directly; other clients such as redis may need a thin wrapper whose methods match these signatures.

Configuration options:

Push NotificationsCopy Anchor Link

Register webhooks to receive task event notifications instead of polling or holding open SSE connections. When a task's status or artifacts change, the server POSTs the event to all registered webhook URLs.

Registering WebhooksCopy Anchor Link

const client = new A2AClient('https://remote-agent.example.com');

const task = await client.sendMessage({
  role: 'user',
  parts: [{ type: 'text', text: 'Run a long analysis' }],
});

const config = await client.createPushNotification(task.id, {
  webhookUrl: 'https://my-app.com/webhooks/a2a',
  authenticationInfo: {
    scheme: 'bearer',
    credentials: { token: 'my-webhook-secret' },
  },
});

Managing WebhooksCopy Anchor Link

const configs = await client.listPushNotifications(task.id);

const single = await client.getPushNotification(task.id, config.id!);

await client.deletePushNotification(task.id, config.id!);

Server ConfigurationCopy Anchor Link

Enable push notifications by providing a PushNotificationStore:

import { A2AServer, InMemoryPushNotificationStore } from '@cogitator-ai/a2a';

const a2aServer = new A2AServer({
  agents: { researcher },
  cogitator,
  pushNotificationStore: new InMemoryPushNotificationStore(),
});

The server automatically sends A2AStreamEvent payloads to registered webhooks on status and artifact updates. Authentication schemes supported: bearer (token), apiKey (key, optional headerName), basic (username, password), oauth2 (accessToken). Configs can only be created for existing tasks.

Unless allowPrivateUrls: true, delivery refuses loopback, private, link-local, CGNAT, multicast and IPv4-mapped IPv6 targets, validates the address the socket actually connects to (no DNS rebinding) and does not follow redirects.

Agent Card SigningCopy Anchor Link

Sign Agent Cards with HMAC-SHA256 to let clients verify card integrity:

Server-SideCopy Anchor Link

Configure cardSigning on the server to automatically sign all Agent Cards:

const a2aServer = new A2AServer({
  agents: { researcher },
  cogitator,
  cardSigning: {
    algorithm: 'hmac-sha256',
    secret: process.env.CARD_SIGNING_SECRET!,
  },
});

Client-Side VerificationCopy Anchor Link

const client = new A2AClient('https://remote-agent.example.com');
const isValid = await client.verifyAgentCard(process.env.CARD_SIGNING_SECRET!);

Manual Signing and VerificationCopy Anchor Link

import { signAgentCard, verifyAgentCardSignature } from '@cogitator-ai/a2a';

const signed = signAgentCard(card, { secret: 'shared-secret' });
console.log(signed.signature); // "hmac-sha256:a1b2c3..."

const valid = verifyAgentCardSignature(signed, 'shared-secret');

Extended Agent CardCopy Anchor Link

The agent/extendedCard method provides an authenticated endpoint that returns additional agent details not included in the public Agent Card -- rate limits, pricing, extended skills, and custom metadata.

Server ConfigurationCopy Anchor Link

const a2aServer: A2AServer = new A2AServer({
  agents: { researcher },
  cogitator,
  extendedCardGenerator: (agentName) => ({
    ...a2aServer.getAgentCard(agentName),
    extendedSkills: [
      {
        id: 'deep_research',
        name: 'deep_research',
        description: 'Multi-source deep research',
        inputModes: ['text/plain'],
        outputModes: ['text/plain', 'application/json'],
      },
    ],
    rateLimit: { requestsPerMinute: 60 },
    pricing: { model: 'per-request', details: '$0.01 per task' },
    metadata: { version: '2.1.0', region: 'us-east-1' },
  }),
});

Fetching from ClientCopy Anchor Link

const client = new A2AClient('https://remote-agent.example.com', {
  headers: { Authorization: 'Bearer my-token' },
});

const extendedCard = await client.extendedAgentCard();
console.log(extendedCard.rateLimit);
console.log(extendedCard.pricing);
console.log(extendedCard.extendedSkills);

The Agent Card advertises this capability via capabilities.extendedAgentCard: true.

Framework AdaptersCopy Anchor Link

All adapters stream only for message/stream (an Accept: text/event-stream header alone does not switch message/send to SSE) and answer JSON-RPC notifications with HTTP 204 and no body. Structurally invalid JSON-RPC requests get -32600 Invalid Request; only unparseable JSON gets -32700 Parse error.

Errors that are neither A2A errors nor CogitatorErrors (in parsing, authentication, routing, streams or agent runs) are logged on the server and answered as -32603 Internal error, without their text; a CogitatorError is answered as -32603 Internal error: <message>. A task that fails that way gets the status message Internal error, and so does the failed status event of a stream.

ExpressCopy Anchor Link

import { a2aExpress } from '@cogitator-ai/a2a/express';
app.use(a2aExpress(a2aServer));

HonoCopy Anchor Link

import { a2aHono } from '@cogitator-ai/a2a/hono';
app.route('/', a2aHono(a2aServer));

FastifyCopy Anchor Link

import { a2aFastify } from '@cogitator-ai/a2a/fastify';
fastify.register(a2aFastify(a2aServer));

KoaCopy Anchor Link

The Koa middleware reads ctx.request.body, so put a JSON body parser in front of it:

import bodyParser from 'koa-bodyparser';
import { a2aKoa } from '@cogitator-ai/a2a/koa';

app.use(bodyParser());
app.use(a2aKoa(a2aServer));

Next.jsCopy Anchor Link

a2aNext returns route handlers: GET answers with the Agent Card and POST with JSON-RPC. Export them from the two routes clients expect:

// app/.well-known/agent.json/route.ts → GET
// app/a2a/route.ts → POST (the route at basePath)
import { a2aNext } from '@cogitator-ai/a2a/next';

export const { GET, POST } = a2aNext(a2aServer);

Task StoreCopy Anchor Link

Tasks are stored in memory by default. InMemoryTaskStore keeps up to maxSize tasks (default 10,000) and evicts the oldest finished task first:

import { A2AServer, InMemoryTaskStore } from '@cogitator-ai/a2a';

const a2aServer = new A2AServer({
  agents: { researcher },
  cogitator,
  taskStore: new InMemoryTaskStore({ maxSize: 50_000 }),
});

For production, use RedisTaskStore or implement the TaskStore interface (create, get, update, list, delete). Your list must honour filter.visibleTo: when it is set, return only tasks owned by that user and tasks without an owner (null means a caller without a user), or tasks/list leaks other users' tasks.

AuthenticationCopy Anchor Link

const a2aServer = new A2AServer({
  agents: { researcher },
  cogitator,
  auth: {
    type: 'bearer',
    validate: async (token) => {
      return token === process.env.A2A_SECRET;
    },
  },
});

validate receives the credential and returns true, false or a caller. Every adapter extracts the credential (Authorization: Bearer ..., or the x-api-key header / headerName for type: 'apiKey') with a2aServer.getAuthToken(); a missing or rejected credential answers a JSON-RPC unauthorized error. The Agent Card itself stays public and advertises the scheme in securitySchemes.

One caller per userCopy Anchor Link

When several users or tenants share the server, return who the caller is instead of true:

auth: {
  type: 'bearer',
  validate: async (token) => {
    const session = await sessions.verify(token);
    return session ? { userId: session.userId } : false;
  },
},

Each task then belongs to the user who created it: another user's tasks/get, tasks/cancel, push-notification calls and attempts to continue the task get Task not found, tasks/list returns only their own tasks (plus tasks without an owner), and a contextId holding someone else's tasks is refused. Runs carry the userId, so the agent's threads and memory are scoped to the user as well (multi-user). The owner is kept in the task's metadata on the server and never sent to clients. Returning true admits the caller without a user: such callers share the tasks that have no owner.

Clients include credentials via headers:

const client = new A2AClient('https://remote-agent.example.com', {
  headers: {
    Authorization: 'Bearer my-secret-token',
  },
});

A2A Protocol MethodsCopy Anchor Link

Task StatesCopy Anchor Link