# A2A Protocol (Agent-to-Agent)

URL: https://cogitator.app/docs/integrations/a2a

> Native implementation of Google's A2A Protocol v0.3 for cross-framework agent interoperability.

## Overview

The `@cogitator-ai/a2a` package implements [Google's A2A Protocol v0.3](https://a2a-protocol.org) — 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

```bash
pnpm add @cogitator-ai/a2a
```

## A2AServer

### Exposing Agents

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

```typescript
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.

```typescript
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 Cards

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

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

```json
{
  "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 Agents

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

```typescript
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:

```typescript
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).

## A2AClient

### Connecting to a Remote Agent

```typescript
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 Messages

```typescript
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);
  }
}
```

### Streaming

```typescript
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 Configuration

| Option                   | Effect                                                                  |
| ------------------------ | ----------------------------------------------------------------------- |
| `blocking: false`        | Return the task immediately in `working` state                          |
| `historyLength`          | Return only the last N history messages (`0` = none); also on `getTask` |
| `acceptedOutputModes`    | Only return artifacts with these MIME types                             |
| `timeout`                | Maximum agent run time in ms                                            |
| `pushNotificationConfig` | Register a webhook before the task starts executing                     |

```typescript
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 Management

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

await client.cancelTask('task_abc123');
```

## asTool() Bridge

Wrap any remote A2A agent as a local Cogitator tool:

```typescript
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 Conversations

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-Side

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

```typescript
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-Side

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 Tasks

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

```typescript
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:

| Field       | Type        | Description                    |
| ----------- | ----------- | ------------------------------ |
| `contextId` | `string`    | Filter by conversation context |
| `state`     | `TaskState` | Filter by task state           |
| `limit`     | `number`    | Max number of tasks to return  |
| `offset`    | `number`    | Number of tasks to skip        |

## Token Streaming

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

```typescript
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:

| Event Type        | Description                             |
| ----------------- | --------------------------------------- |
| `status-update`   | Task state changed (working, completed) |
| `artifact-update` | New artifact produced by the agent      |
| `token`           | Individual token from LLM output        |

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

## RedisTaskStore

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

```typescript
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:

| Option      | Type              | Default     | Description                          |
| ----------- | ----------------- | ----------- | ------------------------------------ |
| `client`    | `RedisClientLike` | required    | Redis client instance                |
| `keyPrefix` | `string`          | `a2a:task:` | Key prefix for task storage          |
| `ttl`       | `number`          | none        | TTL in seconds (uses `SETEX` if set) |

## Push Notifications

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 Webhooks

```typescript
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 Webhooks

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

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

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

### Server Configuration

Enable push notifications by providing a `PushNotificationStore`:

```typescript
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 Signing

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

### Server-Side

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

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

### Client-Side Verification

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

### Manual Signing and Verification

```typescript
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 Card

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 Configuration

```typescript
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 Client

```typescript
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 Adapters

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 `CogitatorError`s (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.

### Express

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

### Hono

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

### Fastify

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

### Koa

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

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

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

### Next.js

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

```typescript
// 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 Store

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

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

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

For production, use [`RedisTaskStore`](#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.

## Authentication

```typescript
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 user

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

```typescript
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](/docs/advanced/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:

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

## A2A Protocol Methods

| Method                          | Description                                                     |
| ------------------------------- | --------------------------------------------------------------- |
| `message/send`                  | Send a message (blocking, or background with `blocking: false`) |
| `message/stream`                | Send a message with SSE streaming (incl. token events)          |
| `tasks/get`                     | Retrieve a task by ID                                           |
| `tasks/cancel`                  | Cancel a running task                                           |
| `tasks/list`                    | List tasks with filtering and pagination                        |
| `tasks/pushNotification/create` | Register a webhook for task events                              |
| `tasks/pushNotification/get`    | Get a push notification config                                  |
| `tasks/pushNotification/list`   | List push notification configs for a task                       |
| `tasks/pushNotification/delete` | Remove a push notification config                               |
| `agent/extendedCard`            | Fetch extended Agent Card (authenticated)                       |

## Task States

| State            | Description                |
| ---------------- | -------------------------- |
| `working`        | Agent is processing        |
| `completed`      | Task finished successfully |
| `failed`         | Task encountered an error  |
| `input-required` | Agent needs more input     |
| `canceled`       | Task was canceled          |
| `rejected`       | Server rejected the task   |