Merge pull request #1 from qixinbo/feature/a2a-production

feat: support a2a mode
2026-04-01 20:49:57 +08:00
parent 9952af198a 86447049a9
commit f073339f7a
12 changed files with 3092 additions and 26 deletions
@@ -186,6 +186,183 @@ python main.py
 ### 6. 初始账号配置 👤
 系统首次注册的用户将自动成为管理员。您可以在登录页面直接点击“注册”按钮创建您的管理员账号（例如：用户名 `admin`，密码 `admin`），随后即可登录并管理项目、数据源和用户。

+### 7. A2A 模式使用指南 🤖
+
+A2A（Agent2Agent）用于让 DataClaw 把任务委托给远端 Agent，并保持任务可跟踪（状态流、产物流、取消、重试）。
+
+#### 7.1 在前端启用 A2A（推荐）
+
+1. 进入 **Skills** 页面，切到 **A2A** 标签页。  
+2. 点击新增远端 Agent，填写：
+   - `name`: 远端 Agent 名称
+   - `base_url`: 远端 DataClaw/A2A 网关地址（如 `https://agent-b.example.com`）
+   - `auth_scheme`: `none` 或 `bearer`
+   - `auth_token`: 当 `auth_scheme=bearer` 时填写
+3. 点击健康检查，确认 `healthy=true`。  
+4. 回到聊天页，开启 **A2A Mode**，选择 `route_mode` 与目标 Agent 后发送问题。  
+5. 在消息卡片与 A2A 任务面板中查看 `SUBMITTED/WORKING/COMPLETED/FAILED` 等状态，可执行取消与重试。
+
+`route_mode` 建议：
+- `auto`: 按项目灰度配置与策略自动决策
+- `local`: 强制本地执行
+- `a2a`: 强制远端 A2A 执行（需选远端 Agent）
+- `a2a_first`: 先远端，失败按回退链执行
+- `local_first`: 先本地，按需再回退
+
+#### 7.2 API 示例（生产常用）
+
+以下示例假设服务地址为 `http://127.0.0.1:8000`，并已获取登录令牌 `${TOKEN}`。
+
+**示例 1：查看本机 A2A Agent Card**
+
+```bash
+curl -H "Authorization: Bearer ${TOKEN}" \
+  http://127.0.0.1:8000/api/v1/a2a/agent-card
+```
+
+**示例 2：注册远端 Agent**
+
+```bash
+curl -X POST http://127.0.0.1:8000/api/v1/a2a/remote-agents \
+  -H "Authorization: Bearer ${TOKEN}" \
+  -H "Content-Type: application/json" \
+  -d '{
+    "project_id": 1,
+    "name": "Agent-B",
+    "base_url": "https://agent-b.example.com",
+    "auth_scheme": "bearer",
+    "auth_token": "remote-agent-token"
+  }'
+```
+
+**示例 3：以 A2A 优先模式发起任务**
+
+```bash
+curl -X POST http://127.0.0.1:8000/api/v1/a2a/messages/send \
+  -H "Authorization: Bearer ${TOKEN}" \
+  -H "Content-Type: application/json" \
+  -d '{
+    "project_id": 1,
+    "message": "请分析最近30天订单转化趋势并给出建议",
+    "session_id": "chat:demo-a2a",
+    "remote_agent_id": 3,
+    "route_mode": "a2a_first",
+    "fallback_chain": ["a2a", "local", "mcp"],
+    "idempotency_key": "demo-a2a-001"
+  }'
+```
+
+返回结果中的 `task.id` 可用于订阅与管理任务。
+
+**示例 4：订阅任务流（SSE）**
+
+```bash
+curl -N -H "Authorization: Bearer ${TOKEN}" \
+  http://127.0.0.1:8000/api/v1/a2a/tasks/<task_id>/subscribe
+```
+
+你会收到类似事件：
+- `TaskStatusUpdateEvent`（状态变更）
+- `TaskArtifactUpdateEvent`（产物更新）
+- `done`（流结束）
+
+**示例 5：取消任务**
+
+```bash
+curl -X POST -H "Authorization: Bearer ${TOKEN}" \
+  http://127.0.0.1:8000/api/v1/a2a/tasks/<task_id>/cancel
+```
+
+**示例 6：为任务配置 Webhook 回调（离线接收结果）**
+
+```bash
+curl -X POST http://127.0.0.1:8000/api/v1/a2a/tasks/<task_id>/webhooks \
+  -H "Authorization: Bearer ${TOKEN}" \
+  -H "Content-Type: application/json" \
+  -d '{
+    "target_url": "https://your-system.example.com/a2a/webhook",
+    "secret": "your-webhook-secret",
+    "auth_header": "Bearer your-internal-token"
+  }'
+```
+
+#### 7.3 一个完整实战流程
+
+场景：你有一个“本地数据分析 Agent”，还接入了“外部行业知识 Agent-B”。
+
+1. 在 Skills -> A2A 注册 `Agent-B` 并完成健康检查。  
+2. 在聊天页开启 A2A，选择 `route_mode=a2a_first`。  
+3. 输入问题：  
+   `请结合外部行业知识与本地销量数据，生成本季度增长策略。`  
+4. 系统先尝试委托给 `Agent-B`；若远端异常则按回退链降级到本地/MCP。  
+5. 在任务面板查看状态流与最终产物，必要时可取消或重试。  
+
+生产建议：
+- 对业务侧请求始终传 `idempotency_key`，避免重复任务。
+- 为长任务配置 webhook，避免客户端断线丢失进度。
+- 在项目级 rollout 配置灰度比例，先小流量启用 A2A 再全量放开。
+
+#### 7.4 本地调试 A2A（双实例联调）
+
+推荐在本机同时启动两个后端实例：
+- 实例 A（调用方）：`http://127.0.0.1:8000`
+- 实例 B（被调用远端 Agent）：`http://127.0.0.1:8001`
+
+分别用两个终端启动（建议使用不同 `DATA_ROOT`，避免数据目录冲突）：
+
+```bash
+# 终端1：实例 A
+cd backend
+source .venv/bin/activate
+DATA_ROOT=/tmp/dataclaw-a uvicorn main:app --reload --port 8000
+```
+
+```bash
+# 终端2：实例 B
+cd backend
+source .venv/bin/activate
+DATA_ROOT=/tmp/dataclaw-b uvicorn main:app --reload --port 8001
+```
+
+然后在两个实例分别注册并登录，拿到 token：
+
+```bash
+# 分别注册（每个实例首次注册用户会成为管理员）
+curl -X POST http://127.0.0.1:8000/api/v1/auth/register \
+  -H "Content-Type: application/json" \
+  -d '{"username":"admin_a","email":"a@test.com","password":"admin12345"}'
+
+curl -X POST http://127.0.0.1:8001/api/v1/auth/register \
+  -H "Content-Type: application/json" \
+  -d '{"username":"admin_b","email":"b@test.com","password":"admin12345"}'
+
+# 登录并保存 token
+TOKEN_A=$(curl -s -X POST http://127.0.0.1:8000/api/v1/auth/login \
+  -H "Content-Type: application/x-www-form-urlencoded" \
+  -d "username=admin_a&password=admin12345" | jq -r '.access_token')
+
+TOKEN_B=$(curl -s -X POST http://127.0.0.1:8001/api/v1/auth/login \
+  -H "Content-Type: application/x-www-form-urlencoded" \
+  -d "username=admin_b&password=admin12345" | jq -r '.access_token')
+```
+
+最后，在实例 A 中把实例 B 注册成远端 Agent（`auth_token` 使用 `TOKEN_B`）：
+
+```bash
+curl -X POST http://127.0.0.1:8000/api/v1/a2a/remote-agents \
+  -H "Authorization: Bearer ${TOKEN_A}" \
+  -H "Content-Type: application/json" \
+  -d "{
+    \"project_id\": 1,
+    \"name\": \"local-agent-b\",
+    \"base_url\": \"http://127.0.0.1:8001\",
+    \"auth_scheme\": \"bearer\",
+    \"auth_token\": \"${TOKEN_B}\"
+  }"
+```
+
+完成后即可在实例 A 发起 A2A 任务、订阅任务流、取消任务，完成本地端到端联调。
+
 ***

 ## 🔌 数据源配置说明
@@ -186,6 +186,134 @@ Frontend setup:
 ### 4. Initial Account Setup 👤
 The first user to register in the system will automatically be granted admin privileges. You can simply click the "Register" button on the login page to create your admin account (e.g., Username: `admin`, Password: `admin`), and then log in to manage projects, data sources, and users.

+### 5. A2A Mode Guide 🤖
+
+A2A (Agent2Agent) lets DataClaw delegate tasks to remote agents with full task lifecycle controls (status stream, artifact stream, cancel, retry).
+
+#### 5.1 Enable A2A in UI (Recommended)
+
+1. Open **Skills** page and switch to the **A2A** tab.
+2. Add a remote agent with:
+   - `name`
+   - `base_url` (for example `https://agent-b.example.com`)
+   - `auth_scheme` (`none` or `bearer`)
+   - `auth_token` (required when `auth_scheme=bearer`)
+3. Run health check and confirm `healthy=true`.
+4. Go to Chat, enable **A2A Mode**, choose `route_mode` and remote agent, then send your prompt.
+5. Track task states in Chat (`SUBMITTED/WORKING/COMPLETED/FAILED`) and use cancel/retry when needed.
+
+`route_mode` quick reference:
+- `auto`: Use project rollout policy and routing strategy
+- `local`: Force local execution
+- `a2a`: Force remote A2A execution
+- `a2a_first`: Try remote first, then fallback chain
+- `local_first`: Try local first
+
+#### 5.2 API Examples
+
+Assume service URL is `http://127.0.0.1:8000` and your bearer token is `${TOKEN}`.
+
+```bash
+# 1) Get local Agent Card
+curl -H "Authorization: Bearer ${TOKEN}" \
+  http://127.0.0.1:8000/api/v1/a2a/agent-card
+
+# 2) Register remote agent
+curl -X POST http://127.0.0.1:8000/api/v1/a2a/remote-agents \
+  -H "Authorization: Bearer ${TOKEN}" \
+  -H "Content-Type: application/json" \
+  -d '{
+    "project_id": 1,
+    "name": "Agent-B",
+    "base_url": "https://agent-b.example.com",
+    "auth_scheme": "bearer",
+    "auth_token": "remote-agent-token"
+  }'
+
+# 3) Send task with a2a_first route
+curl -X POST http://127.0.0.1:8000/api/v1/a2a/messages/send \
+  -H "Authorization: Bearer ${TOKEN}" \
+  -H "Content-Type: application/json" \
+  -d '{
+    "project_id": 1,
+    "message": "Analyze order conversion trend for last 30 days and propose actions",
+    "session_id": "chat:demo-a2a",
+    "remote_agent_id": 3,
+    "route_mode": "a2a_first",
+    "fallback_chain": ["a2a", "local", "mcp"],
+    "idempotency_key": "demo-a2a-001"
+  }'
+
+# 4) Subscribe task stream
+curl -N -H "Authorization: Bearer ${TOKEN}" \
+  http://127.0.0.1:8000/api/v1/a2a/tasks/<task_id>/subscribe
+
+# 5) Cancel task
+curl -X POST -H "Authorization: Bearer ${TOKEN}" \
+  http://127.0.0.1:8000/api/v1/a2a/tasks/<task_id>/cancel
+```
+
+#### 5.3 Local Debugging for A2A (Two-Instance Setup)
+
+Use two local backend instances:
+- Instance A (caller): `http://127.0.0.1:8000`
+- Instance B (remote agent): `http://127.0.0.1:8001`
+
+Run them in two terminals:
+
+```bash
+# Terminal 1 - Instance A
+cd backend
+source .venv/bin/activate
+DATA_ROOT=/tmp/dataclaw-a uvicorn main:app --reload --port 8000
+```
+
+```bash
+# Terminal 2 - Instance B
+cd backend
+source .venv/bin/activate
+DATA_ROOT=/tmp/dataclaw-b uvicorn main:app --reload --port 8001
+```
+
+Create/login users and fetch tokens:
+
+```bash
+# Register (first user becomes admin) - run once per instance
+curl -X POST http://127.0.0.1:8000/api/v1/auth/register \
+  -H "Content-Type: application/json" \
+  -d '{"username":"admin_a","email":"a@test.com","password":"admin12345"}'
+
+curl -X POST http://127.0.0.1:8001/api/v1/auth/register \
+  -H "Content-Type: application/json" \
+  -d '{"username":"admin_b","email":"b@test.com","password":"admin12345"}'
+
+# Login and keep tokens
+TOKEN_A=$(curl -s -X POST http://127.0.0.1:8000/api/v1/auth/login \
+  -H "Content-Type: application/x-www-form-urlencoded" \
+  -d "username=admin_a&password=admin12345" | jq -r '.access_token')
+
+TOKEN_B=$(curl -s -X POST http://127.0.0.1:8001/api/v1/auth/login \
+  -H "Content-Type: application/x-www-form-urlencoded" \
+  -d "username=admin_b&password=admin12345" | jq -r '.access_token')
+```
+
+Then register B as remote agent in A, using `TOKEN_B` as `auth_token`:
+
+```bash
+curl -X POST http://127.0.0.1:8000/api/v1/a2a/remote-agents \
+  -H "Authorization: Bearer ${TOKEN_A}" \
+  -H "Content-Type: application/json" \
+  -d "{
+    \"project_id\": 1,
+    \"name\": \"local-agent-b\",
+    \"base_url\": \"http://127.0.0.1:8001\",
+    \"auth_scheme\": \"bearer\",
+    \"auth_token\": \"${TOKEN_B}\"
+  }"
+```
+
+Finally, send/subscribe/cancel tasks from A. This validates the complete local A2A flow.
+
 ***

 ## 🔌 Data Source Configuration Guide
@@ -0,0 +1,891 @@
+import asyncio
+import json
+from datetime import datetime
+from typing import Any, Dict, List, Literal, Optional, Tuple
+
+import httpx
+from fastapi import APIRouter, Depends, Header, HTTPException, Query, status
+from fastapi.responses import StreamingResponse
+from pydantic import BaseModel, Field
+from sqlalchemy.orm import Session
+
+from app.core.nanobot import nanobot_service
+from app.core.security import CurrentUser, get_current_user
+from app.database import SessionLocal, get_db
+from app.models.a2a import (
+    A2AAuditLog,
+    A2AProjectConfig,
+    A2ARemoteAgent,
+    A2ATask,
+    A2ATaskEvent,
+    A2ATaskWebhook,
+    A2AWebhookDelivery,
+)
+from app.models.project import Project
+from app.services.a2a_service import _json_dumps, _json_loads, a2a_runtime
+from app.trace import build_error_attributes, trace_service
+
+router = APIRouter(prefix="/a2a", tags=["a2a"])
+
+SUPPORTED_PROTOCOL_VERSION = "1.0"
+SUPPORTED_CAPABILITIES = ["streaming", "push", "task_management", "subscribe"]
+SUPPORTED_AUTH = ["bearer", "shared_secret", "none"]
+
+
+def _mask_error(message: str) -> str:
+    if not message:
+        return "internal_error"
+    return "request_failed"
+
+
+class AgentCardResponse(BaseModel):
+    name: str
+    protocol_version: str
+    capabilities: List[str]
+    endpoints: Dict[str, str]
+    auth: List[str]
+
+
+class RemoteAgentCreate(BaseModel):
+    project_id: int
+    name: str = Field(min_length=1, max_length=120)
+    base_url: str = Field(min_length=1, max_length=500)
+    auth_scheme: Literal["none", "bearer"] = "none"
+    auth_token: Optional[str] = None
+
+
+class RemoteAgentUpdate(BaseModel):
+    name: Optional[str] = None
+    base_url: Optional[str] = None
+    auth_scheme: Optional[Literal["none", "bearer"]] = None
+    auth_token: Optional[str] = None
+
+
+class RemoteAgentView(BaseModel):
+    id: int
+    project_id: int
+    name: str
+    base_url: str
+    auth_scheme: str
+    protocol_version: Optional[str] = None
+    capabilities: List[str] = []
+    healthy: bool
+    failure_count: int
+    circuit_open_until: Optional[datetime] = None
+    card_fetched_at: Optional[datetime] = None
+
+
+class SendMessageRequest(BaseModel):
+    project_id: int
+    message: str = Field(min_length=1)
+    session_id: str = "api:a2a"
+    remote_agent_id: Optional[int] = None
+    route_mode: Literal["auto", "local", "a2a", "a2a_first", "local_first", "mcp_first"] = "auto"
+    fallback_chain: Optional[List[Literal["a2a", "local", "mcp"]]] = None
+    idempotency_key: Optional[str] = None
+    metadata: Optional[Dict[str, Any]] = None
+
+
+class TaskView(BaseModel):
+    id: str
+    project_id: int
+    source: str
+    state: str
+    remote_agent_id: Optional[int] = None
+    input_text: str
+    output_text: Optional[str] = None
+    error_message: Optional[str] = None
+    compatibility_mode: bool
+    metadata: Dict[str, Any]
+    created_at: datetime
+    updated_at: datetime
+    finished_at: Optional[datetime] = None
+
+
+class CancelTaskResponse(BaseModel):
+    task_id: str
+    state: str
+
+
+class TaskWebhookCreate(BaseModel):
+    target_url: str = Field(min_length=1, max_length=500)
+    secret: Optional[str] = None
+    auth_header: Optional[str] = None
+
+
+class TaskWebhookView(BaseModel):
+    id: int
+    task_id: str
+    target_url: str
+    enabled: bool
+    created_at: datetime
+    updated_at: datetime
+
+
+class RolloutConfigView(BaseModel):
+    project_id: int
+    canary_enabled: bool
+    canary_percent: int
+    rollback_to_local: bool
+    compatibility_mode: bool
+    dual_event_write: bool
+    route_mode_default: str
+    fallback_chain: List[str]
+    alert_thresholds: Dict[str, Any]
+
+
+class RolloutConfigUpdate(BaseModel):
+    canary_enabled: Optional[bool] = None
+    canary_percent: Optional[int] = Field(default=None, ge=0, le=100)
+    rollback_to_local: Optional[bool] = None
+    compatibility_mode: Optional[bool] = None
+    dual_event_write: Optional[bool] = None
+    route_mode_default: Optional[str] = None
+    fallback_chain: Optional[List[str]] = None
+    alert_thresholds: Optional[Dict[str, Any]] = None
+
+
+def _ensure_project_access(db: Session, project_id: int, user: CurrentUser) -> Project:
+    project = db.query(Project).filter(Project.id == project_id).first()
+    if not project:
+        raise HTTPException(status_code=404, detail="Project not found")
+    if not user.is_admin and project.owner_id != user.id:
+        raise HTTPException(status_code=404, detail="Resource not found")
+    return project
+
+
+def _ensure_task_access(db: Session, task_id: str, user: CurrentUser) -> A2ATask:
+    task = db.query(A2ATask).filter(A2ATask.id == task_id).first()
+    if not task:
+        raise HTTPException(status_code=404, detail="Task not found")
+    if not user.is_admin and task.tenant_id != user.id:
+        raise HTTPException(status_code=404, detail="Task not found")
+    return task
+
+
+def _ensure_agent_access(db: Session, agent_id: int, user: CurrentUser) -> A2ARemoteAgent:
+    agent = db.query(A2ARemoteAgent).filter(A2ARemoteAgent.id == agent_id).first()
+    if not agent:
+        raise HTTPException(status_code=404, detail="Remote agent not found")
+    project = _ensure_project_access(db, agent.project_id, user)
+    if not project:
+        raise HTTPException(status_code=404, detail="Remote agent not found")
+    return agent
+
+
+def _task_to_view(task: A2ATask) -> TaskView:
+    return TaskView(
+        id=task.id,
+        project_id=task.project_id,
+        source=task.source,
+        state=task.state,
+        remote_agent_id=task.remote_agent_id,
+        input_text=task.input_text,
+        output_text=task.output_text,
+        error_message=task.error_message,
+        compatibility_mode=task.compatibility_mode,
+        metadata=_json_loads(task.metadata_json, {}),
+        created_at=task.created_at,
+        updated_at=task.updated_at,
+        finished_at=task.finished_at,
+    )
+
+
+def _agent_to_view(agent: A2ARemoteAgent) -> RemoteAgentView:
+    return RemoteAgentView(
+        id=agent.id,
+        project_id=agent.project_id,
+        name=agent.name,
+        base_url=agent.base_url,
+        auth_scheme=agent.auth_scheme,
+        protocol_version=agent.protocol_version,
+        capabilities=_json_loads(agent.capabilities_json, []),
+        healthy=bool(agent.healthy),
+        failure_count=int(agent.failure_count or 0),
+        circuit_open_until=agent.circuit_open_until,
+        card_fetched_at=agent.card_fetched_at,
+    )
+
+
+def _build_status_event(task: A2ATask, *, compatibility_mode: bool, dual_event_write: bool) -> Dict[str, Any]:
+    payload: Dict[str, Any] = {
+        "type": "TaskStatusUpdateEvent",
+        "task_id": task.id,
+        "task_status": task.state,
+        "timestamp": datetime.utcnow().isoformat(),
+        "source": task.source,
+    }
+    if compatibility_mode or dual_event_write:
+        payload.update(
+            {
+                "event": "task_status",
+                "status": task.state,
+                "taskId": task.id,
+            }
+        )
+    return payload
+
+
+def _build_artifact_event(task_id: str, content: str, *, compatibility_mode: bool, dual_event_write: bool) -> Dict[str, Any]:
+    payload: Dict[str, Any] = {
+        "type": "TaskArtifactUpdateEvent",
+        "task_id": task_id,
+        "artifact": {"content": content},
+        "timestamp": datetime.utcnow().isoformat(),
+    }
+    if compatibility_mode or dual_event_write:
+        payload.update(
+            {
+                "event": "task_output",
+                "taskId": task_id,
+                "output": content,
+            }
+        )
+    return payload
+
+
+async def _delegate_to_remote(task: A2ATask, agent: A2ARemoteAgent, message: str) -> Tuple[str, Dict[str, Any]]:
+    headers: Dict[str, str] = {}
+    if agent.auth_scheme == "bearer" and agent.auth_token:
+        headers["Authorization"] = f"Bearer {agent.auth_token}"
+    payload = {
+        "project_id": task.project_id,
+        "message": message,
+        "session_id": f"a2a-delegate:{task.id}",
+        "idempotency_key": task.idempotency_key,
+        "route_mode": "local_first",
+        "metadata": {"delegated_by": "dataclaw", "task_id": task.id},
+    }
+    url = f"{agent.base_url.rstrip('/')}/api/v1/a2a/messages/send"
+    async with httpx.AsyncClient(timeout=25.0, verify=True) as client:
+        resp = await client.post(url, json=payload, headers=headers)
+    if resp.status_code >= 400:
+        raise RuntimeError(f"remote_http_{resp.status_code}")
+    body = resp.json()
+    content = ""
+    if isinstance(body, dict):
+        task_obj = body.get("task") or {}
+        content = str(task_obj.get("output_text") or body.get("message") or "")
+    return content, body
+
+
+async def _run_task(task_id: str, request: SendMessageRequest, tenant_id: int) -> None:
+    db = SessionLocal()
+    try:
+        task = db.query(A2ATask).filter(A2ATask.id == task_id).first()
+        if not task:
+            return
+        config = a2a_runtime.get_project_config(db, task.project_id, tenant_id)
+        if task.state in {"CANCELED", "REJECTED"}:
+            return
+        with trace_service.start_span("a2a.task.execute", attributes={"task_id": task.id, "project_id": task.project_id, "source": task.source}) as span:
+            start_ts = datetime.utcnow().timestamp()
+            try:
+                task = a2a_runtime.transition_task(db, task, to_state="WORKING")
+                status_event = _build_status_event(task, compatibility_mode=config.compatibility_mode, dual_event_write=config.dual_event_write)
+                status_row = a2a_runtime.append_event(db, task, "TaskStatusUpdateEvent", status_event)
+                await a2a_runtime.publish(task.id, status_event)
+                await a2a_runtime.notify_webhooks(db, task, status_row)
+
+                if task.source == "a2a" and task.remote_agent_id:
+                    agent = db.query(A2ARemoteAgent).filter(A2ARemoteAgent.id == task.remote_agent_id).first()
+                    if not agent:
+                        raise RuntimeError("remote_agent_missing")
+                    response_text, metadata = await _delegate_to_remote(task, agent, request.message)
+                else:
+                    response_text = await nanobot_service.process_message(
+                        request.message,
+                        session_id=f"a2a-task:{task.id}",
+                        project_id=task.project_id,
+                    )
+                    metadata = {"executor": "local"}
+                artifact_event_payload = _build_artifact_event(task.id, response_text or "", compatibility_mode=config.compatibility_mode, dual_event_write=config.dual_event_write)
+                artifact_event = a2a_runtime.append_event(db, task, "TaskArtifactUpdateEvent", artifact_event_payload)
+                await a2a_runtime.publish(task.id, artifact_event_payload)
+                await a2a_runtime.notify_webhooks(db, task, artifact_event)
+                task = a2a_runtime.transition_task(
+                    db,
+                    task,
+                    to_state="COMPLETED",
+                    output_text=response_text or "",
+                    metadata=metadata,
+                )
+                done_event = _build_status_event(task, compatibility_mode=config.compatibility_mode, dual_event_write=config.dual_event_write)
+                done_row = a2a_runtime.append_event(db, task, "TaskStatusUpdateEvent", done_event)
+                await a2a_runtime.publish(task.id, done_event)
+                await a2a_runtime.notify_webhooks(db, task, done_row)
+                elapsed = (datetime.utcnow().timestamp() - start_ts) * 1000
+                await a2a_runtime.metrics.observe_latency("a2a.execute", elapsed)
+            except Exception as exc:
+                span.set_attributes(build_error_attributes(exc, stage="a2a_task_execute"))
+                await a2a_runtime.metrics.incr("a2a.requests.error")
+                task = db.query(A2ATask).filter(A2ATask.id == task.id).first()
+                if task and task.state not in {"COMPLETED", "FAILED", "CANCELED", "REJECTED"}:
+                    task = a2a_runtime.transition_task(db, task, to_state="FAILED", error_message=_json_dumps({"message": _mask_error(str(exc))}))
+                    fail_event = _build_status_event(task, compatibility_mode=task.compatibility_mode, dual_event_write=True)
+                    fail_row = a2a_runtime.append_event(db, task, "TaskStatusUpdateEvent", fail_event)
+                    await a2a_runtime.publish(task.id, fail_event)
+                    await a2a_runtime.notify_webhooks(db, task, fail_row)
+    finally:
+        db.close()
+
+
+@router.get("/agent-card", response_model=AgentCardResponse)
+def get_agent_card() -> AgentCardResponse:
+    return AgentCardResponse(
+        name="DataClaw A2A Gateway",
+        protocol_version=SUPPORTED_PROTOCOL_VERSION,
+        capabilities=SUPPORTED_CAPABILITIES,
+        endpoints={
+            "send_message": "/api/v1/a2a/messages/send",
+            "send_streaming_message": "/api/v1/a2a/messages/stream",
+            "get_task": "/api/v1/a2a/tasks/{task_id}",
+            "list_tasks": "/api/v1/a2a/tasks",
+            "cancel_task": "/api/v1/a2a/tasks/{task_id}/cancel",
+            "subscribe_task": "/api/v1/a2a/tasks/{task_id}/subscribe",
+        },
+        auth=SUPPORTED_AUTH,
+    )
+
+
+@router.get("/remote-agents", response_model=List[RemoteAgentView])
+def list_remote_agents(
+    project_id: Optional[int] = Query(default=None),
+    db: Session = Depends(get_db),
+    current_user: CurrentUser = Depends(get_current_user),
+) -> List[RemoteAgentView]:
+    query = db.query(A2ARemoteAgent)
+    if project_id is not None:
+        _ensure_project_access(db, project_id, current_user)
+        query = query.filter(A2ARemoteAgent.project_id == project_id)
+    if not current_user.is_admin:
+        owned_ids = [p.id for p in db.query(Project).filter(Project.owner_id == current_user.id).all()]
+        if not owned_ids:
+            return []
+        query = query.filter(A2ARemoteAgent.project_id.in_(owned_ids))
+    return [_agent_to_view(item) for item in query.order_by(A2ARemoteAgent.id.desc()).all()]
+
+
+@router.post("/remote-agents", response_model=RemoteAgentView, status_code=status.HTTP_201_CREATED)
+async def create_remote_agent(
+    payload: RemoteAgentCreate,
+    db: Session = Depends(get_db),
+    current_user: CurrentUser = Depends(get_current_user),
+) -> RemoteAgentView:
+    _ensure_project_access(db, payload.project_id, current_user)
+    item = A2ARemoteAgent(
+        project_id=payload.project_id,
+        name=payload.name.strip(),
+        base_url=payload.base_url.strip().rstrip("/"),
+        auth_scheme=payload.auth_scheme,
+        auth_token=payload.auth_token,
+        created_by=current_user.id,
+    )
+    db.add(item)
+    db.commit()
+    db.refresh(item)
+    try:
+        await a2a_runtime.fetch_agent_card(db, item)
+    except Exception:
+        pass
+    a2a_runtime.record_audit(
+        db,
+        actor_user_id=current_user.id,
+        action="create_remote_agent",
+        target_type="remote_agent",
+        target_id=str(item.id),
+        result="ok",
+        project_id=item.project_id,
+    )
+    return _agent_to_view(item)
+
+
+@router.put("/remote-agents/{agent_id}", response_model=RemoteAgentView)
+async def update_remote_agent(
+    agent_id: int,
+    payload: RemoteAgentUpdate,
+    db: Session = Depends(get_db),
+    current_user: CurrentUser = Depends(get_current_user),
+) -> RemoteAgentView:
+    item = _ensure_agent_access(db, agent_id, current_user)
+    update_data = payload.model_dump(exclude_unset=True)
+    for key, value in update_data.items():
+        setattr(item, key, value)
+    if item.base_url:
+        item.base_url = item.base_url.rstrip("/")
+    db.add(item)
+    db.commit()
+    db.refresh(item)
+    try:
+        await a2a_runtime.fetch_agent_card(db, item)
+    except Exception:
+        pass
+    a2a_runtime.record_audit(
+        db,
+        actor_user_id=current_user.id,
+        action="update_remote_agent",
+        target_type="remote_agent",
+        target_id=str(item.id),
+        result="ok",
+        project_id=item.project_id,
+    )
+    return _agent_to_view(item)
+
+
+@router.delete("/remote-agents/{agent_id}")
+def delete_remote_agent(
+    agent_id: int,
+    db: Session = Depends(get_db),
+    current_user: CurrentUser = Depends(get_current_user),
+) -> Dict[str, str]:
+    item = _ensure_agent_access(db, agent_id, current_user)
+    db.delete(item)
+    db.commit()
+    a2a_runtime.record_audit(
+        db,
+        actor_user_id=current_user.id,
+        action="delete_remote_agent",
+        target_type="remote_agent",
+        target_id=str(agent_id),
+        result="ok",
+        project_id=item.project_id,
+    )
+    return {"status": "success"}
+
+
+@router.post("/remote-agents/{agent_id}/refresh-card", response_model=RemoteAgentView)
+async def refresh_remote_agent_card(
+    agent_id: int,
+    db: Session = Depends(get_db),
+    current_user: CurrentUser = Depends(get_current_user),
+) -> RemoteAgentView:
+    item = _ensure_agent_access(db, agent_id, current_user)
+    try:
+        card = await a2a_runtime.fetch_agent_card(db, item)
+    except Exception as exc:
+        a2a_runtime.record_audit(
+            db,
+            actor_user_id=current_user.id,
+            action="refresh_remote_agent_card",
+            target_type="remote_agent",
+            target_id=str(agent_id),
+            result="failed",
+            project_id=item.project_id,
+            detail={"error": str(exc)},
+        )
+        raise HTTPException(status_code=502, detail="Remote card fetch failed")
+    version = str(card.get("protocol_version") or "")
+    if version and version.split(".")[0] != SUPPORTED_PROTOCOL_VERSION.split(".")[0]:
+        raise HTTPException(status_code=400, detail="Protocol version incompatible")
+    return _agent_to_view(item)
+
+
+@router.post("/remote-agents/{agent_id}/health-check")
+async def health_check_remote_agent(
+    agent_id: int,
+    db: Session = Depends(get_db),
+    current_user: CurrentUser = Depends(get_current_user),
+) -> Dict[str, Any]:
+    item = _ensure_agent_access(db, agent_id, current_user)
+    try:
+        await a2a_runtime.fetch_agent_card(db, item, timeout_s=5.0)
+        return {"healthy": True, "failure_count": item.failure_count}
+    except Exception:
+        return {"healthy": False, "failure_count": item.failure_count}
+
+
+@router.post("/messages/send")
+async def send_message(
+    request: SendMessageRequest,
+    x_a2a_token: Optional[str] = Header(default=None),
+    db: Session = Depends(get_db),
+    current_user: CurrentUser = Depends(get_current_user),
+) -> Dict[str, Any]:
+    _ensure_project_access(db, request.project_id, current_user)
+    config = a2a_runtime.get_project_config(db, request.project_id, current_user.id)
+    route = a2a_runtime.resolve_route(
+        project_config=config,
+        session_id=request.session_id,
+        requested_mode=request.route_mode,
+        requested_fallback=request.fallback_chain,
+    )
+    selected_source = "local"
+    remote_agent_id = None
+    if route.selected == "a2a" and request.remote_agent_id:
+        agent = _ensure_agent_access(db, request.remote_agent_id, current_user)
+        if not agent.healthy and config.rollback_to_local:
+            selected_source = "local"
+        else:
+            selected_source = "a2a"
+            remote_agent_id = agent.id
+    task = a2a_runtime.create_task(
+        db,
+        project_id=request.project_id,
+        tenant_id=current_user.id,
+        source=selected_source,
+        input_text=request.message,
+        idempotency_key=request.idempotency_key,
+        remote_agent_id=remote_agent_id,
+        compatibility_mode=config.compatibility_mode,
+        metadata={"route": route.model_dump() if hasattr(route, "model_dump") else route.__dict__, "token_present": bool(x_a2a_token), "request_metadata": request.metadata or {}},
+    )
+    event_payload = _build_status_event(task, compatibility_mode=config.compatibility_mode, dual_event_write=config.dual_event_write)
+    event_row = a2a_runtime.append_event(db, task, "TaskStatusUpdateEvent", event_payload)
+    await a2a_runtime.publish(task.id, event_payload)
+    await a2a_runtime.notify_webhooks(db, task, event_row)
+    asyncio.create_task(_run_task(task.id, request, current_user.id))
+    await a2a_runtime.metrics.incr("a2a.requests.total")
+    a2a_runtime.record_audit(
+        db,
+        actor_user_id=current_user.id,
+        action="send_message",
+        target_type="task",
+        target_id=task.id,
+        result="accepted",
+        project_id=task.project_id,
+        task_id=task.id,
+    )
+    return {"task": _task_to_view(task).model_dump(), "routing": route.__dict__}
+
+
+@router.post("/messages/stream")
+async def send_streaming_message(
+    request: SendMessageRequest,
+    db: Session = Depends(get_db),
+    current_user: CurrentUser = Depends(get_current_user),
+) -> StreamingResponse:
+    response = await send_message(request=request, x_a2a_token=None, db=db, current_user=current_user)
+    task_id = response["task"]["id"]
+
+    async def event_generator():
+        history = (
+            db.query(A2ATaskEvent)
+            .filter(A2ATaskEvent.task_id == task_id)
+            .order_by(A2ATaskEvent.id.asc())
+            .all()
+        )
+        for item in history:
+            payload = _json_loads(item.payload_json, {})
+            yield f"data: {json.dumps(payload, ensure_ascii=False)}\n\n"
+        async for payload in a2a_runtime.subscribe(task_id):
+            yield f"data: {json.dumps(payload, ensure_ascii=False)}\n\n"
+            if payload.get("task_status") in {"COMPLETED", "FAILED", "CANCELED", "REJECTED"}:
+                break
+        yield f"data: {json.dumps({'type': 'done'}, ensure_ascii=False)}\n\n"
+
+    return StreamingResponse(
+        event_generator(),
+        media_type="text/event-stream",
+        headers={"Cache-Control": "no-cache", "Connection": "keep-alive", "X-Accel-Buffering": "no"},
+    )
+
+
+@router.get("/tasks/{task_id}", response_model=TaskView)
+def get_task(
+    task_id: str,
+    db: Session = Depends(get_db),
+    current_user: CurrentUser = Depends(get_current_user),
+) -> TaskView:
+    task = _ensure_task_access(db, task_id, current_user)
+    return _task_to_view(task)
+
+
+@router.get("/tasks", response_model=List[TaskView])
+def list_tasks(
+    project_id: Optional[int] = Query(default=None),
+    state: Optional[str] = Query(default=None),
+    skip: int = Query(default=0, ge=0),
+    limit: int = Query(default=50, ge=1, le=200),
+    db: Session = Depends(get_db),
+    current_user: CurrentUser = Depends(get_current_user),
+) -> List[TaskView]:
+    query = db.query(A2ATask)
+    if not current_user.is_admin:
+        query = query.filter(A2ATask.tenant_id == current_user.id)
+    if project_id is not None:
+        _ensure_project_access(db, project_id, current_user)
+        query = query.filter(A2ATask.project_id == project_id)
+    if state:
+        query = query.filter(A2ATask.state == state)
+    tasks = query.order_by(A2ATask.created_at.desc()).offset(skip).limit(limit).all()
+    return [_task_to_view(item) for item in tasks]
+
+
+@router.post("/tasks/{task_id}/cancel", response_model=CancelTaskResponse)
+async def cancel_task(
+    task_id: str,
+    db: Session = Depends(get_db),
+    current_user: CurrentUser = Depends(get_current_user),
+) -> CancelTaskResponse:
+    task = _ensure_task_access(db, task_id, current_user)
+    if task.state in {"COMPLETED", "FAILED", "CANCELED", "REJECTED"}:
+        return CancelTaskResponse(task_id=task.id, state=task.state)
+    try:
+        task = a2a_runtime.transition_task(db, task, to_state="CANCELED")
+    except ValueError:
+        raise HTTPException(status_code=409, detail="Task state transition conflict")
+    config = a2a_runtime.get_project_config(db, task.project_id, current_user.id)
+    payload = _build_status_event(task, compatibility_mode=config.compatibility_mode, dual_event_write=config.dual_event_write)
+    row = a2a_runtime.append_event(db, task, "TaskStatusUpdateEvent", payload)
+    await a2a_runtime.publish(task.id, payload)
+    await a2a_runtime.notify_webhooks(db, task, row)
+    a2a_runtime.record_audit(
+        db,
+        actor_user_id=current_user.id,
+        action="cancel_task",
+        target_type="task",
+        target_id=task.id,
+        result="ok",
+        project_id=task.project_id,
+        task_id=task.id,
+    )
+    return CancelTaskResponse(task_id=task.id, state=task.state)
+
+
+@router.get("/tasks/{task_id}/subscribe")
+async def subscribe_task(
+    task_id: str,
+    db: Session = Depends(get_db),
+    current_user: CurrentUser = Depends(get_current_user),
+) -> StreamingResponse:
+    task = _ensure_task_access(db, task_id, current_user)
+    initial_events = (
+        db.query(A2ATaskEvent)
+        .filter(A2ATaskEvent.task_id == task.id)
+        .order_by(A2ATaskEvent.id.asc())
+        .all()
+    )
+
+    async def event_generator():
+        for event in initial_events:
+            payload = _json_loads(event.payload_json, {})
+            yield f"data: {json.dumps(payload, ensure_ascii=False)}\n\n"
+        if task.state in {"COMPLETED", "FAILED", "CANCELED", "REJECTED"}:
+            yield f"data: {json.dumps({'type': 'done'}, ensure_ascii=False)}\n\n"
+            return
+        async for payload in a2a_runtime.subscribe(task.id):
+            yield f"data: {json.dumps(payload, ensure_ascii=False)}\n\n"
+            if payload.get("task_status") in {"COMPLETED", "FAILED", "CANCELED", "REJECTED"}:
+                break
+        yield f"data: {json.dumps({'type': 'done'}, ensure_ascii=False)}\n\n"
+
+    return StreamingResponse(
+        event_generator(),
+        media_type="text/event-stream",
+        headers={"Cache-Control": "no-cache", "Connection": "keep-alive", "X-Accel-Buffering": "no"},
+    )
+
+
+@router.get("/tasks/{task_id}/webhooks", response_model=List[TaskWebhookView])
+def list_task_webhooks(
+    task_id: str,
+    db: Session = Depends(get_db),
+    current_user: CurrentUser = Depends(get_current_user),
+) -> List[TaskWebhookView]:
+    task = _ensure_task_access(db, task_id, current_user)
+    items = db.query(A2ATaskWebhook).filter(A2ATaskWebhook.task_id == task.id).order_by(A2ATaskWebhook.id.desc()).all()
+    return [
+        TaskWebhookView(
+            id=item.id,
+            task_id=item.task_id,
+            target_url=item.target_url,
+            enabled=item.enabled,
+            created_at=item.created_at,
+            updated_at=item.updated_at,
+        )
+        for item in items
+    ]
+
+
+@router.post("/tasks/{task_id}/webhooks", response_model=TaskWebhookView, status_code=status.HTTP_201_CREATED)
+def create_task_webhook(
+    task_id: str,
+    payload: TaskWebhookCreate,
+    db: Session = Depends(get_db),
+    current_user: CurrentUser = Depends(get_current_user),
+) -> TaskWebhookView:
+    task = _ensure_task_access(db, task_id, current_user)
+    item = A2ATaskWebhook(
+        task_id=task.id,
+        target_url=payload.target_url.strip(),
+        secret=payload.secret,
+        auth_header=payload.auth_header,
+        created_by=current_user.id,
+    )
+    db.add(item)
+    db.commit()
+    db.refresh(item)
+    a2a_runtime.record_audit(
+        db,
+        actor_user_id=current_user.id,
+        action="create_task_webhook",
+        target_type="task_webhook",
+        target_id=str(item.id),
+        result="ok",
+        project_id=task.project_id,
+        task_id=task.id,
+    )
+    return TaskWebhookView(
+        id=item.id,
+        task_id=item.task_id,
+        target_url=item.target_url,
+        enabled=item.enabled,
+        created_at=item.created_at,
+        updated_at=item.updated_at,
+    )
+
+
+@router.delete("/tasks/{task_id}/webhooks/{webhook_id}")
+def delete_task_webhook(
+    task_id: str,
+    webhook_id: int,
+    db: Session = Depends(get_db),
+    current_user: CurrentUser = Depends(get_current_user),
+) -> Dict[str, str]:
+    task = _ensure_task_access(db, task_id, current_user)
+    item = db.query(A2ATaskWebhook).filter(A2ATaskWebhook.id == webhook_id, A2ATaskWebhook.task_id == task.id).first()
+    if not item:
+        raise HTTPException(status_code=404, detail="Webhook not found")
+    db.delete(item)
+    db.commit()
+    return {"status": "success"}
+
+
+@router.post("/webhook-deliveries/{delivery_id}/replay")
+async def replay_delivery(
+    delivery_id: int,
+    db: Session = Depends(get_db),
+    current_user: CurrentUser = Depends(get_current_user),
+) -> Dict[str, Any]:
+    delivery = db.query(A2AWebhookDelivery).filter(A2AWebhookDelivery.id == delivery_id).first()
+    if not delivery:
+        raise HTTPException(status_code=404, detail="Delivery not found")
+    task = _ensure_task_access(db, delivery.task_id, current_user)
+    webhook = db.query(A2ATaskWebhook).filter(A2ATaskWebhook.id == delivery.webhook_id).first()
+    event = db.query(A2ATaskEvent).filter(A2ATaskEvent.id == delivery.event_id).first()
+    if not webhook or not event:
+        raise HTTPException(status_code=404, detail="Delivery dependencies not found")
+    await a2a_runtime._deliver_once(db, webhook, event, delivery)
+    return {"status": delivery.status, "attempt": delivery.attempt, "dead_letter": delivery.dead_letter, "task_id": task.id}
+
+
+@router.get("/metrics")
+async def get_metrics(current_user: CurrentUser = Depends(get_current_user)) -> Dict[str, Any]:
+    if not current_user.is_admin:
+        raise HTTPException(status_code=403, detail="Admin permission required")
+    return await a2a_runtime.metrics.snapshot()
+
+
+@router.get("/projects/{project_id}/rollout", response_model=RolloutConfigView)
+def get_rollout_config(
+    project_id: int,
+    db: Session = Depends(get_db),
+    current_user: CurrentUser = Depends(get_current_user),
+) -> RolloutConfigView:
+    _ensure_project_access(db, project_id, current_user)
+    item = a2a_runtime.get_project_config(db, project_id, current_user.id)
+    return RolloutConfigView(
+        project_id=item.project_id,
+        canary_enabled=item.canary_enabled,
+        canary_percent=item.canary_percent,
+        rollback_to_local=item.rollback_to_local,
+        compatibility_mode=item.compatibility_mode,
+        dual_event_write=item.dual_event_write,
+        route_mode_default=item.route_mode_default,
+        fallback_chain=_json_loads(item.fallback_chain_json, ["local"]),
+        alert_thresholds=_json_loads(item.alert_thresholds_json, {}),
+    )
+
+
+@router.put("/projects/{project_id}/rollout", response_model=RolloutConfigView)
+def update_rollout_config(
+    project_id: int,
+    payload: RolloutConfigUpdate,
+    db: Session = Depends(get_db),
+    current_user: CurrentUser = Depends(get_current_user),
+) -> RolloutConfigView:
+    _ensure_project_access(db, project_id, current_user)
+    item = a2a_runtime.get_project_config(db, project_id, current_user.id)
+    data = payload.model_dump(exclude_unset=True)
+    for key, value in data.items():
+        if key == "fallback_chain":
+            item.fallback_chain_json = _json_dumps(value)
+            continue
+        if key == "alert_thresholds":
+            item.alert_thresholds_json = _json_dumps(value)
+            continue
+        setattr(item, key, value)
+    item.updated_by = current_user.id
+    db.add(item)
+    db.commit()
+    db.refresh(item)
+    a2a_runtime.record_audit(
+        db,
+        actor_user_id=current_user.id,
+        action="update_rollout_config",
+        target_type="project_rollout",
+        target_id=str(project_id),
+        result="ok",
+        project_id=project_id,
+    )
+    return RolloutConfigView(
+        project_id=item.project_id,
+        canary_enabled=item.canary_enabled,
+        canary_percent=item.canary_percent,
+        rollback_to_local=item.rollback_to_local,
+        compatibility_mode=item.compatibility_mode,
+        dual_event_write=item.dual_event_write,
+        route_mode_default=item.route_mode_default,
+        fallback_chain=_json_loads(item.fallback_chain_json, ["local"]),
+        alert_thresholds=_json_loads(item.alert_thresholds_json, {}),
+    )
+
+
+@router.get("/alerts")
+def get_alert_panel(
+    project_id: int,
+    db: Session = Depends(get_db),
+    current_user: CurrentUser = Depends(get_current_user),
+) -> Dict[str, Any]:
+    _ensure_project_access(db, project_id, current_user)
+    config = a2a_runtime.get_project_config(db, project_id, current_user.id)
+    thresholds = _json_loads(config.alert_thresholds_json, {})
+    defaults = {"error_rate": 0.05, "p95_ms": 3000, "retry_rate": 0.2, "circuit_open_rate": 0.05}
+    merged = {**defaults, **thresholds}
+    return {
+        "project_id": project_id,
+        "thresholds": merged,
+        "panel": {"metrics_endpoint": "/api/v1/a2a/metrics", "task_list_endpoint": "/api/v1/a2a/tasks"},
+    }
+
+
+@router.get("/audit-logs")
+def list_audit_logs(
+    project_id: Optional[int] = Query(default=None),
+    skip: int = Query(default=0, ge=0),
+    limit: int = Query(default=100, ge=1, le=500),
+    db: Session = Depends(get_db),
+    current_user: CurrentUser = Depends(get_current_user),
+) -> List[Dict[str, Any]]:
+    query = db.query(A2AAuditLog)
+    if project_id is not None:
+        _ensure_project_access(db, project_id, current_user)
+        query = query.filter(A2AAuditLog.project_id == project_id)
+    elif not current_user.is_admin:
+        query = query.filter(A2AAuditLog.actor_user_id == current_user.id)
+    rows = query.order_by(A2AAuditLog.created_at.desc()).offset(skip).limit(limit).all()
+    return [
+        {
+            "id": row.id,
+            "actor_user_id": row.actor_user_id,
+            "action": row.action,
+            "target_type": row.target_type,
+            "target_id": row.target_id,
+            "project_id": row.project_id,
+            "task_id": row.task_id,
+            "result": row.result,
+            "detail": _json_loads(row.detail_json, {}),
+            "created_at": row.created_at,
+        }
+        for row in rows
+    ]
@@ -0,0 +1,134 @@
+from sqlalchemy import Boolean, Column, DateTime, ForeignKey, Integer, String, Text, func
+from sqlalchemy.orm import relationship
+
+from app.database import Base
+
+
+class A2ARemoteAgent(Base):
+    __tablename__ = "a2a_remote_agents"
+
+    id = Column(Integer, primary_key=True, index=True)
+    project_id = Column(Integer, ForeignKey("projects.id"), nullable=False, index=True)
+    name = Column(String, nullable=False)
+    base_url = Column(String, nullable=False)
+    auth_scheme = Column(String, nullable=False, default="none")
+    auth_token = Column(String, nullable=True)
+    protocol_version = Column(String, nullable=True)
+    capabilities_json = Column(Text, nullable=False, default="[]")
+    card_json = Column(Text, nullable=True)
+    card_fetched_at = Column(DateTime, nullable=True)
+    healthy = Column(Boolean, nullable=False, default=False)
+    failure_count = Column(Integer, nullable=False, default=0)
+    circuit_open_until = Column(DateTime, nullable=True)
+    created_by = Column(Integer, ForeignKey("users.id"), nullable=False, index=True)
+    created_at = Column(DateTime, default=func.now())
+    updated_at = Column(DateTime, default=func.now(), onupdate=func.now())
+
+    project = relationship("Project")
+
+
+class A2ATask(Base):
+    __tablename__ = "a2a_tasks"
+
+    id = Column(String, primary_key=True, index=True)
+    project_id = Column(Integer, ForeignKey("projects.id"), nullable=False, index=True)
+    tenant_id = Column(Integer, ForeignKey("users.id"), nullable=False, index=True)
+    source = Column(String, nullable=False, default="local")
+    remote_agent_id = Column(Integer, ForeignKey("a2a_remote_agents.id"), nullable=True, index=True)
+    idempotency_key = Column(String, nullable=True, index=True)
+    state = Column(String, nullable=False, index=True, default="SUBMITTED")
+    input_text = Column(Text, nullable=False, default="")
+    output_text = Column(Text, nullable=True)
+    error_message = Column(Text, nullable=True)
+    compatibility_mode = Column(Boolean, nullable=False, default=True)
+    metadata_json = Column(Text, nullable=False, default="{}")
+    created_at = Column(DateTime, default=func.now())
+    updated_at = Column(DateTime, default=func.now(), onupdate=func.now())
+    finished_at = Column(DateTime, nullable=True)
+
+    project = relationship("Project")
+    remote_agent = relationship("A2ARemoteAgent")
+
+
+class A2ATaskEvent(Base):
+    __tablename__ = "a2a_task_events"
+
+    id = Column(Integer, primary_key=True, index=True)
+    task_id = Column(String, ForeignKey("a2a_tasks.id", ondelete="CASCADE"), nullable=False, index=True)
+    event_type = Column(String, nullable=False)
+    payload_json = Column(Text, nullable=False, default="{}")
+    created_at = Column(DateTime, default=func.now(), index=True)
+
+    task = relationship("A2ATask")
+
+
+class A2ATaskWebhook(Base):
+    __tablename__ = "a2a_task_webhooks"
+
+    id = Column(Integer, primary_key=True, index=True)
+    task_id = Column(String, ForeignKey("a2a_tasks.id", ondelete="CASCADE"), nullable=False, index=True)
+    target_url = Column(String, nullable=False)
+    secret = Column(String, nullable=True)
+    auth_header = Column(String, nullable=True)
+    enabled = Column(Boolean, nullable=False, default=True)
+    created_by = Column(Integer, ForeignKey("users.id"), nullable=False, index=True)
+    created_at = Column(DateTime, default=func.now())
+    updated_at = Column(DateTime, default=func.now(), onupdate=func.now())
+
+    task = relationship("A2ATask")
+
+
+class A2AWebhookDelivery(Base):
+    __tablename__ = "a2a_webhook_deliveries"
+
+    id = Column(Integer, primary_key=True, index=True)
+    task_id = Column(String, ForeignKey("a2a_tasks.id", ondelete="CASCADE"), nullable=False, index=True)
+    webhook_id = Column(Integer, ForeignKey("a2a_task_webhooks.id", ondelete="CASCADE"), nullable=False, index=True)
+    event_id = Column(Integer, ForeignKey("a2a_task_events.id", ondelete="CASCADE"), nullable=False, index=True)
+    attempt = Column(Integer, nullable=False, default=0)
+    status = Column(String, nullable=False, default="PENDING")
+    response_code = Column(Integer, nullable=True)
+    response_body = Column(Text, nullable=True)
+    error_message = Column(Text, nullable=True)
+    next_retry_at = Column(DateTime, nullable=True)
+    delivered_at = Column(DateTime, nullable=True)
+    dead_letter = Column(Boolean, nullable=False, default=False, index=True)
+    created_at = Column(DateTime, default=func.now())
+    updated_at = Column(DateTime, default=func.now(), onupdate=func.now())
+
+    task = relationship("A2ATask")
+    webhook = relationship("A2ATaskWebhook")
+    event = relationship("A2ATaskEvent")
+
+
+class A2AProjectConfig(Base):
+    __tablename__ = "a2a_project_configs"
+
+    project_id = Column(Integer, ForeignKey("projects.id"), primary_key=True)
+    canary_enabled = Column(Boolean, nullable=False, default=False)
+    canary_percent = Column(Integer, nullable=False, default=0)
+    rollback_to_local = Column(Boolean, nullable=False, default=True)
+    compatibility_mode = Column(Boolean, nullable=False, default=True)
+    dual_event_write = Column(Boolean, nullable=False, default=True)
+    route_mode_default = Column(String, nullable=False, default="local_first")
+    fallback_chain_json = Column(Text, nullable=False, default='["local"]')
+    alert_thresholds_json = Column(Text, nullable=False, default="{}")
+    updated_by = Column(Integer, ForeignKey("users.id"), nullable=False)
+    updated_at = Column(DateTime, default=func.now(), onupdate=func.now())
+
+    project = relationship("Project")
+
+
+class A2AAuditLog(Base):
+    __tablename__ = "a2a_audit_logs"
+
+    id = Column(Integer, primary_key=True, index=True)
+    actor_user_id = Column(Integer, ForeignKey("users.id"), nullable=False, index=True)
+    action = Column(String, nullable=False)
+    target_type = Column(String, nullable=False)
+    target_id = Column(String, nullable=False)
+    project_id = Column(Integer, ForeignKey("projects.id"), nullable=True, index=True)
+    task_id = Column(String, nullable=True, index=True)
+    result = Column(String, nullable=False)
+    detail_json = Column(Text, nullable=False, default="{}")
+    created_at = Column(DateTime, default=func.now(), index=True)
@@ -0,0 +1,384 @@
+from __future__ import annotations
+
+import asyncio
+import hashlib
+import hmac
+import json
+import time
+import uuid
+from collections import defaultdict, deque
+from dataclasses import dataclass
+from datetime import datetime, timedelta, timezone
+from typing import Any, AsyncIterator, Dict, Iterable, List, Optional, Tuple
+
+import httpx
+from sqlalchemy.orm import Session
+
+from app.models.a2a import (
+    A2AAuditLog,
+    A2AProjectConfig,
+    A2ARemoteAgent,
+    A2ATask,
+    A2ATaskEvent,
+    A2ATaskWebhook,
+    A2AWebhookDelivery,
+)
+from app.models.project import Project
+from app.trace import build_error_attributes, trace_service
+
+_STATE_TRANSITIONS = {
+    "SUBMITTED": {"WORKING", "FAILED", "CANCELED", "REJECTED", "AUTH_REQUIRED", "INPUT_REQUIRED", "COMPLETED"},
+    "WORKING": {"COMPLETED", "FAILED", "CANCELED", "INPUT_REQUIRED", "AUTH_REQUIRED"},
+    "INPUT_REQUIRED": {"WORKING", "FAILED", "CANCELED"},
+    "AUTH_REQUIRED": {"WORKING", "FAILED", "CANCELED", "REJECTED"},
+    "REJECTED": set(),
+    "FAILED": set(),
+    "COMPLETED": set(),
+    "CANCELED": set(),
+}
+_TERMINAL_STATES = {"COMPLETED", "FAILED", "CANCELED", "REJECTED"}
+
+
+def _json_loads(raw: Optional[str], default: Any) -> Any:
+    if not raw:
+        return default
+    try:
+        return json.loads(raw)
+    except Exception:
+        return default
+
+
+def _json_dumps(raw: Any) -> str:
+    return json.dumps(raw, ensure_ascii=False)
+
+
+def _utc_now() -> datetime:
+    return datetime.now(timezone.utc)
+
+
+def _mask_error(message: str) -> str:
+    if not message:
+        return "internal_error"
+    return "request_failed"
+
+
+@dataclass
+class A2AResolvedRoute:
+    selected: str
+    fallback_chain: List[str]
+    canary_hit: bool
+    reason: str
+
+
+class A2AMetrics:
+    def __init__(self) -> None:
+        self._lock = asyncio.Lock()
+        self._counters: Dict[str, int] = defaultdict(int)
+        self._latency_ms: Dict[str, deque[float]] = defaultdict(lambda: deque(maxlen=2000))
+
+    async def incr(self, key: str, value: int = 1) -> None:
+        async with self._lock:
+            self._counters[key] += value
+
+    async def observe_latency(self, key: str, elapsed_ms: float) -> None:
+        async with self._lock:
+            self._latency_ms[key].append(float(elapsed_ms))
+
+    async def snapshot(self) -> Dict[str, Any]:
+        async with self._lock:
+            counters = dict(self._counters)
+            p95 = {}
+            for key, values in self._latency_ms.items():
+                series = sorted(values)
+                if not series:
+                    p95[f"{key}.p95_ms"] = 0.0
+                    continue
+                idx = int(0.95 * (len(series) - 1))
+                p95[f"{key}.p95_ms"] = round(series[idx], 2)
+            total = counters.get("a2a.requests.total", 0)
+            errors = counters.get("a2a.requests.error", 0)
+            retries = counters.get("a2a.requests.retry", 0)
+            breakers = counters.get("a2a.circuit.open", 0)
+            return {
+                "counters": counters,
+                "derived": {
+                    "error_rate": round(errors / total, 4) if total else 0.0,
+                    "retry_rate": round(retries / total, 4) if total else 0.0,
+                    "circuit_open_rate": round(breakers / total, 4) if total else 0.0,
+                },
+                "latency": p95,
+            }
+
+
+class A2ARuntime:
+    def __init__(self) -> None:
+        self._subscribers: Dict[str, List[asyncio.Queue[Dict[str, Any]]]] = defaultdict(list)
+        self.metrics = A2AMetrics()
+        self.protocol_version = "1.0"
+        self._circuit_state: Dict[int, datetime] = {}
+
+    async def publish(self, task_id: str, event: Dict[str, Any]) -> None:
+        queues = list(self._subscribers.get(task_id, []))
+        for queue in queues:
+            await queue.put(event)
+
+    async def subscribe(self, task_id: str) -> AsyncIterator[Dict[str, Any]]:
+        queue: asyncio.Queue[Dict[str, Any]] = asyncio.Queue(maxsize=200)
+        self._subscribers[task_id].append(queue)
+        try:
+            while True:
+                payload = await queue.get()
+                yield payload
+        finally:
+            self._subscribers[task_id] = [q for q in self._subscribers.get(task_id, []) if q is not queue]
+            if not self._subscribers[task_id]:
+                self._subscribers.pop(task_id, None)
+
+    def get_project_config(self, db: Session, project_id: int, user_id: int) -> A2AProjectConfig:
+        item = db.query(A2AProjectConfig).filter(A2AProjectConfig.project_id == project_id).first()
+        if item:
+            return item
+        config = A2AProjectConfig(project_id=project_id, updated_by=user_id)
+        db.add(config)
+        db.commit()
+        db.refresh(config)
+        return config
+
+    def resolve_route(self, *, project_config: A2AProjectConfig, session_id: str, requested_mode: str, requested_fallback: Optional[List[str]]) -> A2AResolvedRoute:
+        selected = requested_mode or project_config.route_mode_default or "local_first"
+        fallback = requested_fallback or _json_loads(project_config.fallback_chain_json, ["local"])
+        fallback_chain = [item for item in fallback if item in {"a2a", "local", "mcp"}]
+        if not fallback_chain:
+            fallback_chain = ["local"]
+        canary_hit = False
+        if project_config.canary_enabled and project_config.canary_percent > 0:
+            digest = hashlib.sha256(f"{project_config.project_id}:{session_id}".encode()).hexdigest()
+            bucket = int(digest[:8], 16) % 100
+            canary_hit = bucket < project_config.canary_percent
+        if selected in {"a2a_first", "a2a"} and not canary_hit:
+            return A2AResolvedRoute(
+                selected="local",
+                fallback_chain=fallback_chain,
+                canary_hit=False,
+                reason="canary_not_hit_fallback_local",
+            )
+        if selected in {"a2a_first", "a2a"}:
+            return A2AResolvedRoute(selected="a2a", fallback_chain=fallback_chain, canary_hit=canary_hit, reason="a2a_selected")
+        if selected in {"mcp_first", "mcp"}:
+            return A2AResolvedRoute(selected="mcp", fallback_chain=fallback_chain, canary_hit=canary_hit, reason="mcp_selected")
+        return A2AResolvedRoute(selected="local", fallback_chain=fallback_chain, canary_hit=canary_hit, reason="local_selected")
+
+    def can_transition(self, from_state: str, to_state: str) -> bool:
+        if from_state == to_state:
+            return True
+        return to_state in _STATE_TRANSITIONS.get(from_state, set())
+
+    def create_task(
+        self,
+        db: Session,
+        *,
+        project_id: int,
+        tenant_id: int,
+        source: str,
+        input_text: str,
+        idempotency_key: Optional[str],
+        remote_agent_id: Optional[int],
+        compatibility_mode: bool,
+        metadata: Optional[Dict[str, Any]] = None,
+    ) -> A2ATask:
+        if idempotency_key:
+            existing = (
+                db.query(A2ATask)
+                .filter(
+                    A2ATask.project_id == project_id,
+                    A2ATask.tenant_id == tenant_id,
+                    A2ATask.idempotency_key == idempotency_key,
+                )
+                .first()
+            )
+            if existing:
+                return existing
+        task = A2ATask(
+            id=f"task_{uuid.uuid4().hex}",
+            project_id=project_id,
+            tenant_id=tenant_id,
+            source=source,
+            remote_agent_id=remote_agent_id,
+            state="SUBMITTED",
+            input_text=input_text,
+            idempotency_key=idempotency_key,
+            compatibility_mode=compatibility_mode,
+            metadata_json=_json_dumps(metadata or {}),
+        )
+        db.add(task)
+        db.commit()
+        db.refresh(task)
+        return task
+
+    def append_event(self, db: Session, task: A2ATask, event_type: str, payload: Dict[str, Any]) -> A2ATaskEvent:
+        event = A2ATaskEvent(task_id=task.id, event_type=event_type, payload_json=_json_dumps(payload))
+        db.add(event)
+        db.commit()
+        db.refresh(event)
+        return event
+
+    def transition_task(
+        self,
+        db: Session,
+        task: A2ATask,
+        *,
+        to_state: str,
+        output_text: Optional[str] = None,
+        error_message: Optional[str] = None,
+        metadata: Optional[Dict[str, Any]] = None,
+    ) -> A2ATask:
+        if not self.can_transition(task.state, to_state):
+            raise ValueError(f"Invalid task transition: {task.state} -> {to_state}")
+        task.state = to_state
+        if output_text is not None:
+            task.output_text = output_text
+        if error_message is not None:
+            task.error_message = error_message
+        if metadata is not None:
+            task.metadata_json = _json_dumps(metadata)
+        if to_state in _TERMINAL_STATES:
+            task.finished_at = _utc_now()
+        db.add(task)
+        db.commit()
+        db.refresh(task)
+        return task
+
+    def record_audit(
+        self,
+        db: Session,
+        *,
+        actor_user_id: int,
+        action: str,
+        target_type: str,
+        target_id: str,
+        result: str,
+        project_id: Optional[int] = None,
+        task_id: Optional[str] = None,
+        detail: Optional[Dict[str, Any]] = None,
+    ) -> None:
+        audit = A2AAuditLog(
+            actor_user_id=actor_user_id,
+            action=action,
+            target_type=target_type,
+            target_id=target_id,
+            result=result,
+            project_id=project_id,
+            task_id=task_id,
+            detail_json=_json_dumps(detail or {}),
+        )
+        db.add(audit)
+        db.commit()
+
+    async def fetch_agent_card(self, db: Session, agent: A2ARemoteAgent, *, timeout_s: float = 10.0) -> Dict[str, Any]:
+        if agent.id in self._circuit_state and self._circuit_state[agent.id] > _utc_now():
+            raise RuntimeError("circuit_open")
+        started = time.perf_counter()
+        await self.metrics.incr("a2a.requests.total")
+        headers = {}
+        if agent.auth_scheme == "bearer" and agent.auth_token:
+            headers["Authorization"] = f"Bearer {agent.auth_token}"
+        url = f"{agent.base_url.rstrip('/')}/api/v1/a2a/agent-card"
+        with trace_service.start_span("a2a.card.fetch", attributes={"agent_id": agent.id, "url": url}) as span:
+            for attempt in range(3):
+                try:
+                    async with httpx.AsyncClient(timeout=timeout_s, verify=True) as client:
+                        resp = await client.get(url, headers=headers)
+                    if resp.status_code >= 400:
+                        raise RuntimeError(f"http_{resp.status_code}")
+                    payload = resp.json()
+                    elapsed_ms = (time.perf_counter() - started) * 1000
+                    await self.metrics.observe_latency("a2a.card.fetch", elapsed_ms)
+                    agent.card_json = _json_dumps(payload)
+                    agent.protocol_version = str(payload.get("protocol_version") or "")
+                    agent.capabilities_json = _json_dumps(payload.get("capabilities") or [])
+                    agent.card_fetched_at = _utc_now()
+                    agent.healthy = True
+                    agent.failure_count = 0
+                    agent.circuit_open_until = None
+                    db.add(agent)
+                    db.commit()
+                    db.refresh(agent)
+                    return payload
+                except Exception as exc:
+                    span.set_attributes(build_error_attributes(exc, stage="a2a_card_fetch"))
+                    await self.metrics.incr("a2a.requests.error")
+                    if attempt < 2:
+                        await self.metrics.incr("a2a.requests.retry")
+                        await asyncio.sleep(0.2 * (2 ** attempt))
+                        continue
+                    agent.failure_count = (agent.failure_count or 0) + 1
+                    if agent.failure_count >= 3:
+                        reopen_at = _utc_now() + timedelta(seconds=90)
+                        agent.circuit_open_until = reopen_at
+                        self._circuit_state[agent.id] = reopen_at
+                        await self.metrics.incr("a2a.circuit.open")
+                    agent.healthy = False
+                    db.add(agent)
+                    db.commit()
+                    raise
+
+    async def notify_webhooks(self, db: Session, task: A2ATask, event: A2ATaskEvent) -> None:
+        webhooks = db.query(A2ATaskWebhook).filter(A2ATaskWebhook.task_id == task.id, A2ATaskWebhook.enabled == True).all()
+        if not webhooks:
+            return
+        for hook in webhooks:
+            delivery = A2AWebhookDelivery(task_id=task.id, webhook_id=hook.id, event_id=event.id, attempt=0, status="PENDING")
+            db.add(delivery)
+            db.commit()
+            db.refresh(delivery)
+            await self._deliver_once(db, hook, event, delivery)
+
+    async def _deliver_once(self, db: Session, hook: A2ATaskWebhook, event: A2ATaskEvent, delivery: A2AWebhookDelivery) -> None:
+        event_payload = _json_loads(event.payload_json, {})
+        request_payload = {
+            "task_id": event.task_id,
+            "event_type": event.event_type,
+            "event_id": event.id,
+            "payload": event_payload,
+        }
+        body = _json_dumps(request_payload).encode("utf-8")
+        for attempt in range(1, 5):
+            delivery.attempt = attempt
+            db.add(delivery)
+            db.commit()
+            headers = {"Content-Type": "application/json", "X-A2A-Event-Id": str(event.id)}
+            if hook.secret:
+                digest = hmac.new(hook.secret.encode("utf-8"), body, hashlib.sha256).hexdigest()
+                headers["X-A2A-Signature"] = f"sha256={digest}"
+            if hook.auth_header:
+                headers["Authorization"] = hook.auth_header
+            try:
+                async with httpx.AsyncClient(timeout=8.0, verify=True) as client:
+                    resp = await client.post(hook.target_url, content=body, headers=headers)
+                delivery.response_code = resp.status_code
+                delivery.response_body = (resp.text or "")[:1000]
+                if 200 <= resp.status_code < 300:
+                    delivery.status = "DELIVERED"
+                    delivery.dead_letter = False
+                    delivery.delivered_at = _utc_now()
+                    db.add(delivery)
+                    db.commit()
+                    return
+                raise RuntimeError(f"http_{resp.status_code}")
+            except Exception as exc:
+                delivery.error_message = str(exc)[:500]
+                if attempt < 4:
+                    delivery.status = "RETRYING"
+                    delivery.next_retry_at = _utc_now() + timedelta(seconds=2 ** attempt)
+                    db.add(delivery)
+                    db.commit()
+                    await asyncio.sleep(2 ** attempt)
+                    continue
+                delivery.status = "FAILED"
+                delivery.dead_letter = True
+                db.add(delivery)
+                db.commit()
+                return
+
+
+a2a_runtime = A2ARuntime()
@@ -23,7 +23,7 @@ import re
 import os
 from datetime import datetime

-from app.api import upload, llm, skills, users, datasources, projects, semantic, mcp, subagents, knowledge, embedding_models, web_search
+from app.api import upload, llm, skills, users, datasources, projects, semantic, mcp, subagents, knowledge, embedding_models, web_search, a2a
 from app.connectors.postgres import postgres_connector
 from app.connectors.clickhouse import clickhouse_connector
 from app.core.artifacts import extract_artifacts
@@ -52,6 +52,15 @@ from app.models.user import User, EmailVerification
 from app.models.project import Project
 from app.models.datasource import DataSource
 from app.models.subagent import Subagent
+from app.models.a2a import (
+    A2ARemoteAgent,
+    A2ATask,
+    A2ATaskEvent,
+    A2ATaskWebhook,
+    A2AWebhookDelivery,
+    A2AProjectConfig,
+    A2AAuditLog,
+)

 app = FastAPI()

@@ -86,6 +95,7 @@ app.include_router(subagents.router, prefix="/api/v1")
 app.include_router(knowledge.router, prefix="/api/v1")
 app.include_router(embedding_models.router, prefix="/api/v1")
 app.include_router(web_search.router, prefix="/api/v1")
+app.include_router(a2a.router, prefix="/api/v1")

 STREAM_DELTA_CHUNK_SIZE = 48
 PREVIEWABLE_TEXT_EXTENSIONS = {
@@ -292,7 +302,9 @@ class ChatRequest(BaseModel):
    source: str = "postgres"
    prefer_sql_chart: bool = False
    file_url: Optional[str] = None
-    route_mode: Literal["auto", "chat", "sql"] = "auto"
+    route_mode: Literal["auto", "chat", "sql", "a2a", "a2a_first", "local_first", "mcp_first"] = "auto"
+    route_fallback_chain: Optional[List[Literal["a2a", "local", "mcp"]]] = None
+    a2a_agent_id: Optional[int] = None
    knowledge_base_id: Optional[str] = None


@@ -0,0 +1,319 @@
+import asyncio
+import sys
+from collections.abc import Generator
+from pathlib import Path
+
+import httpx
+import pytest
+from fastapi.testclient import TestClient
+from sqlalchemy import create_engine
+from sqlalchemy.orm import Session, sessionmaker
+from sqlalchemy.pool import StaticPool
+
+BACKEND_ROOT = Path(__file__).resolve().parents[1]
+REPO_ROOT = BACKEND_ROOT.parent
+NANOBOT_ROOT = REPO_ROOT / "nanobot"
+if str(BACKEND_ROOT) not in sys.path:
+    sys.path.insert(0, str(BACKEND_ROOT))
+if str(NANOBOT_ROOT) not in sys.path:
+    sys.path.insert(0, str(NANOBOT_ROOT))
+
+from app.core.security import CurrentUser, get_current_user
+from app.database import Base, get_db
+from app.models.a2a import A2ARemoteAgent
+from app.models.project import Project
+from app.models.user import User
+from app.services.a2a_service import a2a_runtime
+from main import app
+
+
+def _seed(db: Session) -> tuple[int, str, int, str, int]:
+    owner = User(username="a2a_owner", email="a2a_owner@example.com", hashed_password="x", is_admin=False)
+    other = User(username="a2a_other", email="a2a_other@example.com", hashed_password="x", is_admin=False)
+    db.add(owner)
+    db.add(other)
+    db.commit()
+    db.refresh(owner)
+    db.refresh(other)
+    project = Project(name="a2a_project", description="a2a", owner_id=owner.id)
+    db.add(project)
+    db.commit()
+    db.refresh(project)
+    return owner.id, owner.username, other.id, other.username, project.id
+
+
+def test_a2a_send_list_cancel_and_rollout() -> None:
+    engine = create_engine("sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool)
+    testing_session_local = sessionmaker(autocommit=False, autoflush=False, bind=engine)
+    Base.metadata.create_all(bind=engine)
+    db = testing_session_local()
+    owner_id, owner_username, _, _, project_id = _seed(db)
+    db.close()
+
+    state = {"user": CurrentUser(id=owner_id, username=owner_username, is_admin=False)}
+
+    def override_get_db() -> Generator[Session, None, None]:
+        override_db = testing_session_local()
+        try:
+            yield override_db
+        finally:
+            override_db.close()
+
+    def override_current_user() -> CurrentUser:
+        return state["user"]
+
+    app.dependency_overrides[get_db] = override_get_db
+    app.dependency_overrides[get_current_user] = override_current_user
+    try:
+        client = TestClient(app)
+        send_resp = client.post(
+            "/api/v1/a2a/messages/send",
+            json={
+                "project_id": project_id,
+                "message": "hello a2a",
+                "session_id": "test-a2a-session",
+                "route_mode": "local_first",
+            },
+        )
+        assert send_resp.status_code == 200
+        task_id = send_resp.json()["task"]["id"]
+
+        get_resp = client.get(f"/api/v1/a2a/tasks/{task_id}")
+        assert get_resp.status_code == 200
+        assert get_resp.json()["project_id"] == project_id
+
+        list_resp = client.get("/api/v1/a2a/tasks", params={"project_id": project_id})
+        assert list_resp.status_code == 200
+        assert any(item["id"] == task_id for item in list_resp.json())
+
+        cancel_resp = client.post(f"/api/v1/a2a/tasks/{task_id}/cancel")
+        assert cancel_resp.status_code == 200
+        assert cancel_resp.json()["state"] in {"CANCELED", "COMPLETED", "FAILED"}
+
+        rollout_resp = client.put(
+            f"/api/v1/a2a/projects/{project_id}/rollout",
+            json={"canary_enabled": True, "canary_percent": 30, "route_mode_default": "a2a_first", "fallback_chain": ["a2a", "local"]},
+        )
+        assert rollout_resp.status_code == 200
+        assert rollout_resp.json()["canary_enabled"] is True
+        assert rollout_resp.json()["canary_percent"] == 30
+    finally:
+        app.dependency_overrides.clear()
+        Base.metadata.drop_all(bind=engine)
+        engine.dispose()
+
+
+def test_a2a_task_tenant_isolation() -> None:
+    engine = create_engine("sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool)
+    testing_session_local = sessionmaker(autocommit=False, autoflush=False, bind=engine)
+    Base.metadata.create_all(bind=engine)
+    db = testing_session_local()
+    owner_id, owner_username, other_id, other_username, project_id = _seed(db)
+    db.close()
+
+    state = {"user": CurrentUser(id=owner_id, username=owner_username, is_admin=False)}
+
+    def override_get_db() -> Generator[Session, None, None]:
+        override_db = testing_session_local()
+        try:
+            yield override_db
+        finally:
+            override_db.close()
+
+    def override_current_user() -> CurrentUser:
+        return state["user"]
+
+    app.dependency_overrides[get_db] = override_get_db
+    app.dependency_overrides[get_current_user] = override_current_user
+    try:
+        client = TestClient(app)
+        send_resp = client.post(
+            "/api/v1/a2a/messages/send",
+            json={"project_id": project_id, "message": "tenant isolation", "session_id": "tenant-isolation", "route_mode": "local"},
+        )
+        assert send_resp.status_code == 200
+        task_id = send_resp.json()["task"]["id"]
+
+        state["user"] = CurrentUser(id=other_id, username=other_username, is_admin=False)
+        forbidden_resp = client.get(f"/api/v1/a2a/tasks/{task_id}")
+        assert forbidden_resp.status_code == 404
+    finally:
+        app.dependency_overrides.clear()
+        Base.metadata.drop_all(bind=engine)
+        engine.dispose()
+
+
+def test_a2a_metrics_admin_only() -> None:
+    engine = create_engine("sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool)
+    testing_session_local = sessionmaker(autocommit=False, autoflush=False, bind=engine)
+    Base.metadata.create_all(bind=engine)
+    db = testing_session_local()
+    owner_id, owner_username, _, _, _ = _seed(db)
+    db.close()
+
+    state = {"user": CurrentUser(id=owner_id, username=owner_username, is_admin=False)}
+
+    def override_get_db() -> Generator[Session, None, None]:
+        override_db = testing_session_local()
+        try:
+            yield override_db
+        finally:
+            override_db.close()
+
+    def override_current_user() -> CurrentUser:
+        return state["user"]
+
+    app.dependency_overrides[get_db] = override_get_db
+    app.dependency_overrides[get_current_user] = override_current_user
+    try:
+        client = TestClient(app)
+        denied = client.get("/api/v1/a2a/metrics")
+        assert denied.status_code == 403
+        state["user"] = CurrentUser(id=owner_id, username=owner_username, is_admin=True)
+        ok = client.get("/api/v1/a2a/metrics")
+        assert ok.status_code == 200
+        assert "counters" in ok.json()
+    finally:
+        app.dependency_overrides.clear()
+        Base.metadata.drop_all(bind=engine)
+        engine.dispose()
+
+
+def test_a2a_send_idempotency_key_deduplicates_task() -> None:
+    engine = create_engine("sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool)
+    testing_session_local = sessionmaker(autocommit=False, autoflush=False, bind=engine)
+    Base.metadata.create_all(bind=engine)
+    db = testing_session_local()
+    owner_id, owner_username, _, _, project_id = _seed(db)
+    db.close()
+
+    state = {"user": CurrentUser(id=owner_id, username=owner_username, is_admin=False)}
+
+    def override_get_db() -> Generator[Session, None, None]:
+        override_db = testing_session_local()
+        try:
+            yield override_db
+        finally:
+            override_db.close()
+
+    def override_current_user() -> CurrentUser:
+        return state["user"]
+
+    app.dependency_overrides[get_db] = override_get_db
+    app.dependency_overrides[get_current_user] = override_current_user
+    try:
+        client = TestClient(app)
+        payload = {
+            "project_id": project_id,
+            "message": "dedupe-task",
+            "session_id": "idempotency-session",
+            "route_mode": "local_first",
+            "idempotency_key": "same-key-1",
+        }
+        first_resp = client.post("/api/v1/a2a/messages/send", json=payload)
+        second_resp = client.post("/api/v1/a2a/messages/send", json=payload)
+        assert first_resp.status_code == 200
+        assert second_resp.status_code == 200
+        assert first_resp.json()["task"]["id"] == second_resp.json()["task"]["id"]
+    finally:
+        app.dependency_overrides.clear()
+        Base.metadata.drop_all(bind=engine)
+        engine.dispose()
+
+
+def test_a2a_fetch_agent_card_auth_failure_marks_agent_unhealthy(monkeypatch) -> None:
+    engine = create_engine("sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool)
+    testing_session_local = sessionmaker(autocommit=False, autoflush=False, bind=engine)
+    Base.metadata.create_all(bind=engine)
+    db = testing_session_local()
+    owner_id, _, _, _, project_id = _seed(db)
+    agent = A2ARemoteAgent(
+        project_id=project_id,
+        name="auth-fail-agent",
+        base_url="https://remote.example.com",
+        auth_scheme="bearer",
+        auth_token="bad-token",
+        created_by=owner_id,
+    )
+    db.add(agent)
+    db.commit()
+    db.refresh(agent)
+    a2a_runtime._circuit_state.pop(agent.id, None)
+
+    class _FailResp:
+        status_code = 401
+
+        @staticmethod
+        def json():
+            return {"detail": "unauthorized"}
+
+    class _Client401:
+        def __init__(self, *args, **kwargs):
+            pass
+
+        async def __aenter__(self):
+            return self
+
+        async def __aexit__(self, exc_type, exc, tb):
+            return False
+
+        async def get(self, url, headers=None):
+            return _FailResp()
+
+    monkeypatch.setattr("app.services.a2a_service.httpx.AsyncClient", _Client401)
+
+    with pytest.raises(RuntimeError):
+        asyncio.run(a2a_runtime.fetch_agent_card(db, agent, timeout_s=0.01))
+    db.refresh(agent)
+    assert agent.healthy is False
+    assert agent.failure_count == 1
+
+    Base.metadata.drop_all(bind=engine)
+    db.close()
+    engine.dispose()
+
+
+def test_a2a_fetch_agent_card_remote_unavailable_opens_circuit(monkeypatch) -> None:
+    engine = create_engine("sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool)
+    testing_session_local = sessionmaker(autocommit=False, autoflush=False, bind=engine)
+    Base.metadata.create_all(bind=engine)
+    db = testing_session_local()
+    owner_id, _, _, _, project_id = _seed(db)
+    agent = A2ARemoteAgent(
+        project_id=project_id,
+        name="offline-agent",
+        base_url="https://offline.example.com",
+        auth_scheme="none",
+        created_by=owner_id,
+    )
+    db.add(agent)
+    db.commit()
+    db.refresh(agent)
+    a2a_runtime._circuit_state.pop(agent.id, None)
+
+    class _ClientDown:
+        def __init__(self, *args, **kwargs):
+            pass
+
+        async def __aenter__(self):
+            return self
+
+        async def __aexit__(self, exc_type, exc, tb):
+            return False
+
+        async def get(self, url, headers=None):
+            raise httpx.ConnectError("network down")
+
+    monkeypatch.setattr("app.services.a2a_service.httpx.AsyncClient", _ClientDown)
+
+    for _ in range(3):
+        with pytest.raises(Exception):
+            asyncio.run(a2a_runtime.fetch_agent_card(db, agent, timeout_s=0.01))
+    db.refresh(agent)
+    assert agent.healthy is False
+    assert agent.failure_count == 3
+    assert agent.circuit_open_until is not None
+
+    Base.metadata.drop_all(bind=engine)
+    db.close()
+    engine.dispose()
@@ -0,0 +1,175 @@
+import { api } from "@/lib/api";
+
+export interface A2ARemoteAgent {
+  id: number;
+  project_id: number;
+  name: string;
+  base_url: string;
+  auth_scheme: "none" | "bearer";
+  protocol_version?: string | null;
+  capabilities: string[];
+  healthy: boolean;
+  failure_count: number;
+  circuit_open_until?: string | null;
+  card_fetched_at?: string | null;
+}
+
+export interface A2ATask {
+  id: string;
+  project_id: number;
+  source: string;
+  state: string;
+  remote_agent_id?: number | null;
+  input_text: string;
+  output_text?: string | null;
+  error_message?: string | null;
+  compatibility_mode: boolean;
+  metadata: Record<string, unknown>;
+  created_at: string;
+  updated_at: string;
+  finished_at?: string | null;
+}
+
+export interface A2ASendMessagePayload {
+  project_id: number;
+  message: string;
+  session_id?: string;
+  remote_agent_id?: number;
+  route_mode?: "auto" | "local" | "a2a" | "a2a_first" | "local_first" | "mcp_first";
+  fallback_chain?: Array<"a2a" | "local" | "mcp">;
+  idempotency_key?: string;
+  metadata?: Record<string, unknown>;
+}
+
+export interface A2ASendMessageResponse {
+  task: A2ATask;
+  routing?: {
+    selected?: string;
+    fallback_chain?: string[];
+    canary_hit?: boolean;
+    reason?: string;
+  };
+}
+
+export interface A2ASubscribeEvent {
+  type?: string;
+  event?: string;
+  task_id?: string;
+  task_status?: string;
+  status?: string;
+  artifact?: {
+    content?: string;
+  };
+  output?: string;
+  source?: string;
+  timestamp?: string;
+}
+
+type SubscribeHandler = (event: A2ASubscribeEvent) => void;
+
+const parseSseEvents = (chunk: string): A2ASubscribeEvent[] => {
+  const blocks = chunk.split("\n\n");
+  const events: A2ASubscribeEvent[] = [];
+  for (const block of blocks) {
+    if (!block.trim()) continue;
+    const lines = block.split("\n");
+    const dataLine = lines.find((line) => line.startsWith("data:"));
+    if (!dataLine) continue;
+    const raw = dataLine.slice(5).trim();
+    if (!raw) continue;
+    try {
+      const parsed = JSON.parse(raw) as A2ASubscribeEvent;
+      events.push(parsed);
+    } catch {
+      continue;
+    }
+  }
+  return events;
+};
+
+const getAuthHeaders = (): Record<string, string> => {
+  const token = localStorage.getItem("token");
+  return token ? { Authorization: `Bearer ${token}` } : {};
+};
+
+export const a2aApi = {
+  listRemoteAgents(projectId: number) {
+    return api.get<A2ARemoteAgent[]>(`/api/v1/a2a/remote-agents?project_id=${projectId}`);
+  },
+  createRemoteAgent(payload: {
+    project_id: number;
+    name: string;
+    base_url: string;
+    auth_scheme: "none" | "bearer";
+    auth_token?: string;
+  }) {
+    return api.post<A2ARemoteAgent>("/api/v1/a2a/remote-agents", payload);
+  },
+  updateRemoteAgent(agentId: number, payload: {
+    name?: string;
+    base_url?: string;
+    auth_scheme?: "none" | "bearer";
+    auth_token?: string;
+  }) {
+    return api.put<A2ARemoteAgent>(`/api/v1/a2a/remote-agents/${agentId}`, payload);
+  },
+  deleteRemoteAgent(agentId: number) {
+    return api.delete<{ status: string }>(`/api/v1/a2a/remote-agents/${agentId}`);
+  },
+  refreshRemoteAgentCard(agentId: number) {
+    return api.post<A2ARemoteAgent>(`/api/v1/a2a/remote-agents/${agentId}/refresh-card`, {});
+  },
+  healthCheckRemoteAgent(agentId: number) {
+    return api.post<{ healthy: boolean; failure_count: number }>(`/api/v1/a2a/remote-agents/${agentId}/health-check`, {});
+  },
+  listTasks(projectId: number, state?: string) {
+    const params = new URLSearchParams({ project_id: String(projectId), limit: "100" });
+    if (state && state !== "all") {
+      params.set("state", state);
+    }
+    return api.get<A2ATask[]>(`/api/v1/a2a/tasks?${params.toString()}`);
+  },
+  getTask(taskId: string) {
+    return api.get<A2ATask>(`/api/v1/a2a/tasks/${taskId}`);
+  },
+  cancelTask(taskId: string) {
+    return api.post<{ task_id: string; state: string }>(`/api/v1/a2a/tasks/${taskId}/cancel`, {});
+  },
+  sendMessage(payload: A2ASendMessagePayload) {
+    return api.post<A2ASendMessageResponse>("/api/v1/a2a/messages/send", payload);
+  },
+  async subscribeTask(taskId: string, onEvent: SubscribeHandler, signal?: AbortSignal): Promise<void> {
+    const response = await fetch(`/api/v1/a2a/tasks/${taskId}/subscribe`, {
+      method: "GET",
+      headers: {
+        ...getAuthHeaders(),
+      },
+      signal,
+    });
+    if (!response.ok || !response.body) {
+      throw new Error(`Subscribe failed: ${response.status}`);
+    }
+    const reader = response.body.getReader();
+    const decoder = new TextDecoder("utf-8");
+    let buffer = "";
+    while (true) {
+      const { done, value } = await reader.read();
+      if (done) break;
+      buffer += decoder.decode(value, { stream: true });
+      const splitIndex = buffer.lastIndexOf("\n\n");
+      if (splitIndex === -1) continue;
+      const complete = buffer.slice(0, splitIndex);
+      buffer = buffer.slice(splitIndex + 2);
+      const events = parseSseEvents(complete);
+      for (const event of events) {
+        onEvent(event);
+      }
+    }
+    if (buffer.trim()) {
+      const events = parseSseEvents(buffer);
+      for (const event of events) {
+        onEvent(event);
+      }
+    }
+  },
+};
@@ -1,7 +1,8 @@
 import { useState, useRef, useEffect } from "react";
 import { ScrollArea } from "@/components/ui/scroll-area";
-import { User, Loader2, ArrowUp, ChevronDown, Check, Square, Plus, Database, Wand2, CheckCircle2, Table, XCircle, Settings, ExternalLink, Download, Copy, Mic, X, Compass } from "lucide-react";
+import { User, Loader2, ArrowUp, ChevronDown, Check, Square, Plus, Database, Wand2, CheckCircle2, Table, XCircle, Settings, ExternalLink, Download, Copy, Mic, X, Compass, RotateCcw } from "lucide-react";
 import { api } from "@/lib/api";
+import { a2aApi, type A2ARemoteAgent, type A2ATask, type A2ASubscribeEvent } from "@/api/a2a";
 import { type ChartSpec } from "@/store/visualizationStore";
 import { Popover, PopoverContent, PopoverTrigger } from "@/components/ui/popover";
 import { cn } from "@/lib/utils";
@@ -31,6 +32,12 @@ interface Message {
  };
  artifacts?: MessageArtifact[];
  kbCitations?: KnowledgeCitation[];
+  a2aTaskId?: string;
+  a2aTaskState?: string;
+  a2aRouteMode?: string;
+  a2aRemoteAgentId?: number | null;
+  a2aInputText?: string;
+  a2aError?: string;
 }

 interface MessageViz {
@@ -81,7 +88,7 @@ const splitReportHtml = (content: string): { markdown: string; reportHtml: strin
  return { markdown, reportHtml: reportHtml || null };
 };

-const HTML_FILE_REGEX = /data[\\\/]data[\\\/]([a-zA-Z0-9_\-]+\.html?)/i;
+const HTML_FILE_REGEX = /data[\\/]data[\\/]([a-zA-Z0-9_-]+\.html?)/i;

 const extractExternalReport = (content: string): string | null => {
  if (!content) return null;
@@ -135,13 +142,25 @@ interface SessionData {
    active_data_file?: DataFileContext | null;
    selected_data_source?: string | null;
    selected_knowledge_base_id?: string | null;
-    [key: string]: any;
+    [key: string]: unknown;
  };
-  messages: Array<{
-    role: string;
-    content: string;
-    [key: string]: any;
-  }>;
+  messages: SessionMessage[];
+}
+
+interface SessionMessage {
+  role: string;
+  content: string;
+  tool_calls?: unknown[];
+  viz?: unknown;
+  reasoning_content?: string;
+  usage?: {
+    prompt_tokens: number;
+    completion_tokens: number;
+    total_tokens: number;
+  };
+  artifacts?: unknown;
+  kb_citations?: unknown;
+  [key: string]: unknown;
 }

 const normalizeArtifacts = (raw: unknown): MessageArtifact[] => {
@@ -201,6 +220,12 @@ const normalizeKnowledgeCitations = (raw: unknown): KnowledgeCitation[] => {
  }, []);
 };

+const getErrorMessage = (error: unknown): string => {
+  if (error instanceof Error) return error.message;
+  if (typeof error === "string") return error;
+  return "";
+};
+
 export function ChatInterface() {
  const { t } = useTranslation();
  const [messagesBySession, setMessagesBySession] = useState<Record<string, Message[]>>({});
@@ -222,6 +247,13 @@ export function ChatInterface() {

  const [availableSkills, setAvailableSkills] = useState<Skill[]>([]);
  const [selectedSkillIds, setSelectedSkillIds] = useState<string[]>([]);
+  const [a2aEnabled, setA2aEnabled] = useState(false);
+  const [a2aRouteMode, setA2aRouteMode] = useState<"auto" | "local" | "a2a" | "a2a_first" | "local_first">("auto");
+  const [a2aRemoteAgents, setA2aRemoteAgents] = useState<A2ARemoteAgent[]>([]);
+  const [selectedA2aAgentId, setSelectedA2aAgentId] = useState<string>("");
+  const [a2aTaskStateFilter, setA2aTaskStateFilter] = useState<string>("all");
+  const [a2aTasks, setA2aTasks] = useState<A2ATask[]>([]);
+  const [isA2aTaskLoading, setIsA2aTaskLoading] = useState(false);
  const filteredSlashSkills = slashQuery !== null 
    ? availableSkills.filter(s => s.name.toLowerCase().includes(slashQuery.toLowerCase()))
    : [];
@@ -233,7 +265,7 @@ export function ChatInterface() {
    
    // Remove the slash command from input
    // Match the last occurrence of /query
-    const match = input.match(/(?:^|\s)\/([a-zA-Z0-9_\-]*)$/);
+    const match = input.match(/(?:^|\s)\/([a-zA-Z0-9_-]*)$/);
    if (match && match.index !== undefined) {
       // match[0] includes the leading space if present
       const prefix = input.slice(0, match.index);
@@ -282,7 +314,7 @@ export function ChatInterface() {
    setInput(val);
    
    // Simple slash detection: if the last word starts with /
-    const match = val.match(/(?:^|\s)\/([a-zA-Z0-9_\-]*)$/);
+    const match = val.match(/(?:^|\s)\/([a-zA-Z0-9_-]*)$/);
    if (match) {
      setSlashQuery(match[1]);
      setSlashIndex(0);
@@ -311,11 +343,46 @@ export function ChatInterface() {
  
  const generatingSessionsRef = useRef<Record<string, boolean>>({});
  const abortControllersRef = useRef<Record<string, AbortController>>({});
+  const a2aSubscribeControllersRef = useRef<Record<string, AbortController>>({});
+  const a2aActiveTaskBySessionRef = useRef<Record<string, string>>({});

  // Model selection state
  const [models, setModels] = useState<ModelConfig[]>([]);
  const [selectedModelId, setSelectedModelId] = useState<string>("");

+  const isTerminalA2aState = (state?: string) => {
+    return state ? ["COMPLETED", "FAILED", "CANCELED", "REJECTED"].includes(state) : false;
+  };
+
+  const parseA2aErrorMessage = (raw?: string | null) => {
+    if (!raw) return "";
+    try {
+      const parsed = JSON.parse(raw) as { message?: string };
+      return parsed.message || raw;
+    } catch {
+      return raw;
+    }
+  };
+
+  const fetchA2aAgentsAndTasks = async (projectId: number, stateFilter: string = a2aTaskStateFilter) => {
+    setIsA2aTaskLoading(true);
+    try {
+      const [agents, tasks] = await Promise.all([
+        a2aApi.listRemoteAgents(projectId),
+        a2aApi.listTasks(projectId, stateFilter),
+      ]);
+      setA2aRemoteAgents(agents || []);
+      setA2aTasks(tasks || []);
+      if (selectedA2aAgentId && !(agents || []).some((agent) => String(agent.id) === selectedA2aAgentId)) {
+        setSelectedA2aAgentId("");
+      }
+    } catch (error) {
+      console.error("Failed to fetch A2A agents or tasks", error);
+    } finally {
+      setIsA2aTaskLoading(false);
+    }
+  };
+
  // Listen for model changes from the ProjectSwitcher
  useEffect(() => {
    const handleModelChange = (e: Event) => {
@@ -486,11 +553,15 @@ export function ChatInterface() {
    if (currentProject) {
      fetchDataSources();
      fetchKnowledgeBases();
+      void fetchA2aAgentsAndTasks(currentProject.id, a2aTaskStateFilter);
    } else {
      setAvailableKnowledgeBases([]);
      setSelectedKnowledgeBaseId("");
+      setA2aRemoteAgents([]);
+      setA2aTasks([]);
+      setSelectedA2aAgentId("");
    }
-  }, [currentProject]);
+  }, [currentProject, a2aTaskStateFilter]);

  const fetchDataSources = async () => {
    if (!currentProject) return;
@@ -733,12 +804,179 @@ export function ChatInterface() {
    }
  };

+  const updateA2aMessageByTaskId = (sessionKey: string, taskId: string, updater: (msg: Message) => Message) => {
+    setMessagesForSession(sessionKey, (prev) =>
+      prev.map((msg) => (msg.a2aTaskId === taskId ? updater(msg) : msg))
+    );
+  };
+
+  const syncTaskSnapshotToMessage = (sessionKey: string, task: A2ATask) => {
+    updateA2aMessageByTaskId(sessionKey, task.id, (msg) => ({
+      ...msg,
+      a2aTaskState: task.state,
+      content: task.output_text || msg.content,
+      a2aError: parseA2aErrorMessage(task.error_message),
+      awaitingFirstToken: !isTerminalA2aState(task.state),
+    }));
+  };
+
+  const applyA2aSubscribeEvent = (sessionKey: string, taskId: string, event: A2ASubscribeEvent) => {
+    if (event.type === "TaskStatusUpdateEvent") {
+      const state = event.task_status || event.status || "";
+      updateA2aMessageByTaskId(sessionKey, taskId, (msg) => {
+        const currentLogs = msg.progressLogs || [];
+        const nextLog = state ? `${t('a2aStatus')}: ${state}` : "";
+        const logs = nextLog && currentLogs[currentLogs.length - 1] !== nextLog ? [...currentLogs, nextLog] : currentLogs;
+        return {
+          ...msg,
+          a2aTaskState: state || msg.a2aTaskState,
+          progressLogs: logs,
+          awaitingFirstToken: state ? !isTerminalA2aState(state) : msg.awaitingFirstToken,
+        };
+      });
+      return;
+    }
+    if (event.type === "TaskArtifactUpdateEvent") {
+      const content = event.artifact?.content || event.output || "";
+      if (!content) return;
+      updateA2aMessageByTaskId(sessionKey, taskId, (msg) => ({
+        ...msg,
+        content,
+        awaitingFirstToken: false,
+      }));
+    }
+  };
+
+  const runA2aMessageFlow = async (sessionKey: string, assistantId: string, inputText: string) => {
+    if (!currentProject) return;
+    const payload = {
+      project_id: currentProject.id,
+      message: inputText,
+      session_id: sessionKey,
+      route_mode: a2aRouteMode,
+      ...(selectedA2aAgentId ? { remote_agent_id: Number(selectedA2aAgentId) } : {}),
+      metadata: { from_chat: true },
+    } as const;
+    const response = await a2aApi.sendMessage(payload);
+    const task = response.task;
+    a2aActiveTaskBySessionRef.current[sessionKey] = task.id;
+    setMessagesForSession(sessionKey, (prev) =>
+      prev.map((msg) =>
+        msg.id === assistantId
+          ? {
+              ...msg,
+              a2aTaskId: task.id,
+              a2aTaskState: task.state,
+              a2aRouteMode: a2aRouteMode,
+              a2aRemoteAgentId: task.remote_agent_id || null,
+              a2aInputText: inputText,
+              routeInfo: response.routing?.selected || msg.routeInfo,
+              progressLogs: [...(msg.progressLogs || []), `${t('a2aTaskCreated')}: ${task.id}`],
+            }
+          : msg
+      )
+    );
+    await fetchA2aAgentsAndTasks(currentProject.id, a2aTaskStateFilter);
+    const subscribeController = new AbortController();
+    a2aSubscribeControllersRef.current[task.id] = subscribeController;
+    try {
+      await a2aApi.subscribeTask(
+        task.id,
+        (event) => {
+          applyA2aSubscribeEvent(sessionKey, task.id, event);
+        },
+        subscribeController.signal
+      );
+    } catch (error) {
+      if (!subscribeController.signal.aborted) {
+        console.error("A2A subscribe failed", error);
+      }
+    } finally {
+      delete a2aSubscribeControllersRef.current[task.id];
+      const latestTask = await a2aApi.getTask(task.id).catch(() => null);
+      if (latestTask) {
+        syncTaskSnapshotToMessage(sessionKey, latestTask);
+      }
+      if (currentProject) {
+        await fetchA2aAgentsAndTasks(currentProject.id, a2aTaskStateFilter);
+      }
+    }
+  };
+
+  const handleCancelA2aTask = async (taskId: string) => {
+    try {
+      await a2aApi.cancelTask(taskId);
+      const controller = a2aSubscribeControllersRef.current[taskId];
+      if (controller) {
+        controller.abort();
+        delete a2aSubscribeControllersRef.current[taskId];
+      }
+      if (currentProject) {
+        await fetchA2aAgentsAndTasks(currentProject.id, a2aTaskStateFilter);
+      }
+      setMessagesForSession(activeSessionKey, (prev) =>
+        prev.map((msg) => (msg.a2aTaskId === taskId ? { ...msg, a2aTaskState: "CANCELED", awaitingFirstToken: false } : msg))
+      );
+    } catch (error) {
+      console.error("Failed to cancel A2A task", error);
+    }
+  };
+
+  const handleRetryA2aTask = async (msg: Message) => {
+    if (!msg.a2aInputText) return;
+    const targetSessionKey = activeSessionKey;
+    const newUserMessage: Message = { id: Date.now().toString(), role: "user", content: msg.a2aInputText };
+    setMessagesForSession(targetSessionKey, (prev) => [...prev, newUserMessage]);
+    const assistantId = (Date.now() + 1).toString();
+    setMessagesForSession(targetSessionKey, (prev) => [
+      ...prev,
+      {
+        id: assistantId,
+        role: "assistant",
+        content: "",
+        awaitingFirstToken: true,
+        progressLogs: [t('requestSubmittedRouting')],
+      },
+    ]);
+    setIsLoadingForSession(targetSessionKey, true);
+    generatingSessionsRef.current[targetSessionKey] = true;
+    try {
+      await runA2aMessageFlow(targetSessionKey, assistantId, msg.a2aInputText);
+    } catch (error: unknown) {
+      const errorMessage = getErrorMessage(error) || t('unknownError');
+      setMessagesForSession(targetSessionKey, (prev) =>
+        prev.map((item) =>
+          item.id === assistantId
+            ? {
+                ...item,
+                awaitingFirstToken: false,
+                a2aTaskState: "FAILED",
+                a2aError: errorMessage,
+                content: item.content || `${t('a2aTaskFailed')}: ${errorMessage}`,
+              }
+            : item
+        )
+      );
+    } finally {
+      generatingSessionsRef.current[targetSessionKey] = false;
+      setIsLoadingForSession(targetSessionKey, false);
+      delete a2aActiveTaskBySessionRef.current[targetSessionKey];
+    }
+  };
+
  const renderActiveSelections = () => {
    const hasValidDataSourceSelection = Boolean(selectedDataSource && selectedDataSourceName);
-    if (!hasValidDataSourceSelection && !selectedKnowledgeBaseId) return null;
+    const selectedAgent = a2aRemoteAgents.find((agent) => String(agent.id) === selectedA2aAgentId);
+    if (!hasValidDataSourceSelection && !selectedKnowledgeBaseId && !a2aEnabled) return null;
    return (
      <div className="px-2 pt-2">
        <div className="flex flex-wrap gap-2">
+          {a2aEnabled ? (
+            <div className="px-3 py-1.5 rounded-full text-xs border flex items-center gap-1.5 bg-emerald-50 text-emerald-700 border-emerald-200">
+              <Database className="h-3.5 w-3.5" />
+              {`A2A：${a2aRouteMode}${selectedAgent ? ` · ${selectedAgent.name}` : ""}`}
+            </div>
+          ) : null}
          {hasValidDataSourceSelection ? (
            <div className="px-3 py-1.5 rounded-full text-xs border flex items-center gap-1.5 bg-blue-50 text-blue-700 border-blue-200">
              <Database className="h-3.5 w-3.5" />
@@ -803,6 +1041,58 @@ export function ChatInterface() {
    return (
      <div className="relative group max-w-4xl mx-auto">
        <div className="flex flex-col bg-background rounded-[26px] border border-border shadow-[0_2px_12px_rgba(0,0,0,0.04)] transition-all duration-200">
+          <div className="px-3 pt-2">
+            <div className="flex flex-wrap items-center gap-2 text-xs">
+              <button
+                type="button"
+                className={cn(
+                  "px-2.5 py-1 rounded-full border transition-colors",
+                  a2aEnabled ? "bg-emerald-50 text-emerald-700 border-emerald-200" : "bg-muted text-muted-foreground border-border"
+                )}
+                onClick={() => setA2aEnabled((prev) => !prev)}
+              >
+                {a2aEnabled ? t('a2aModeEnabled') : t('a2aModeDisabled')}
+              </button>
+              {a2aEnabled ? (
+                <>
+                  <select
+                    value={a2aRouteMode}
+                    onChange={(e) => setA2aRouteMode(e.target.value as "auto" | "local" | "a2a" | "a2a_first" | "local_first")}
+                    className="h-7 rounded-md border border-border bg-background px-2 text-xs text-foreground"
+                  >
+                    <option value="auto">auto</option>
+                    <option value="local">local</option>
+                    <option value="a2a">a2a</option>
+                    <option value="a2a_first">a2a_first</option>
+                    <option value="local_first">local_first</option>
+                  </select>
+                  <select
+                    value={selectedA2aAgentId}
+                    onChange={(e) => setSelectedA2aAgentId(e.target.value)}
+                    className="h-7 rounded-md border border-border bg-background px-2 text-xs text-foreground min-w-[160px]"
+                  >
+                    <option value="">{t('autoSelectAgent')}</option>
+                    {a2aRemoteAgents.map((agent) => (
+                      <option key={agent.id} value={String(agent.id)}>
+                        {agent.name} ({agent.healthy ? t('healthy') : t('unhealthy')})
+                      </option>
+                    ))}
+                  </select>
+                  <button
+                    type="button"
+                    className="h-7 px-2 rounded-md border border-border text-muted-foreground hover:text-foreground"
+                    onClick={() => {
+                      if (currentProject) {
+                        void fetchA2aAgentsAndTasks(currentProject.id, a2aTaskStateFilter);
+                      }
+                    }}
+                  >
+                    {t('refresh')}
+                  </button>
+                </>
+              ) : null}
+            </div>
+          </div>
          {renderFileCard()}
          {renderActiveSelections()}
          <div className="flex items-center pl-2 pr-2 py-2">
@@ -1037,9 +1327,15 @@ export function ChatInterface() {
  }, [messages]);

  const handleForceStop = () => {
+    const activeTaskId = a2aActiveTaskBySessionRef.current[activeSessionKey];
+    if (activeTaskId) {
+      void handleCancelA2aTask(activeTaskId);
+      delete a2aActiveTaskBySessionRef.current[activeSessionKey];
+    }
    const controller = abortControllersRef.current[activeSessionKey];
-    if (!controller) return;
-    controller.abort();
+    if (controller) {
+      controller.abort();
+    }
    setIsLoadingForSession(activeSessionKey, false);
    generatingSessionsRef.current[activeSessionKey] = false;
    setMessagesForSession(activeSessionKey, (prev) =>
@@ -1065,6 +1361,49 @@ export function ChatInterface() {
      messagePayload = `[${t('userUploadedFile')}: ${currentAttachedFile.filename}]\n[${t('fileContentSummary')}: ${currentAttachedFile.summary || t('none')}]\n[${t('dataColumns')}: ${currentAttachedFile.columns?.join(", ") || t('none')}]\n[${t('fileDownloadLink')}: ${currentAttachedFile.url}]\n\n${newMessage.content}`;
      setAttachedFile(null);
    }
+
+    if (a2aEnabled && currentProject) {
+      generatingSessionsRef.current[targetSessionKey] = true;
+      setIsLoadingForSession(targetSessionKey, true);
+      const assistantId = (Date.now() + 1).toString();
+      setMessagesForSession(targetSessionKey, (prev) => [
+        ...prev,
+        {
+          id: assistantId,
+          role: "assistant",
+          content: "",
+          awaitingFirstToken: true,
+          progressLogs: [t('requestSubmittedRouting')],
+          a2aRouteMode,
+          a2aRemoteAgentId: selectedA2aAgentId ? Number(selectedA2aAgentId) : null,
+          a2aInputText: messagePayload,
+        },
+      ]);
+      try {
+        await runA2aMessageFlow(targetSessionKey, assistantId, messagePayload);
+      } catch (error: unknown) {
+        const errorMessage = getErrorMessage(error) || t('unknownError');
+        setMessagesForSession(targetSessionKey, (prev) =>
+          prev.map((msg) =>
+            msg.id === assistantId
+              ? {
+                  ...msg,
+                  awaitingFirstToken: false,
+                  a2aTaskState: "FAILED",
+                  a2aError: errorMessage,
+                  content: msg.content || `${t('a2aTaskFailed')}: ${errorMessage}`,
+                }
+              : msg
+          )
+        );
+      } finally {
+        generatingSessionsRef.current[targetSessionKey] = false;
+        setIsLoadingForSession(targetSessionKey, false);
+        delete a2aActiveTaskBySessionRef.current[targetSessionKey];
+        window.dispatchEvent(new Event("nanobot:sessions-changed"));
+      }
+      return;
+    }
    
    const controller = new AbortController();
    abortControllersRef.current[targetSessionKey] = controller;
@@ -1318,8 +1657,9 @@ export function ChatInterface() {
          )
        );
       }
-    } catch (error: any) {
-        if (error?.name === "AbortError" || String(error?.message || "").toLowerCase().includes("aborted")) {
+    } catch (error: unknown) {
+        const errorMessage = getErrorMessage(error);
+        if (errorMessage.toLowerCase().includes("aborted")) {
          setMessagesForSession(targetSessionKey, (prev) =>
            prev.map((msg) =>
              msg.awaitingFirstToken
@@ -1332,7 +1672,7 @@ export function ChatInterface() {
        setMessagesForSession(targetSessionKey, prev => [...prev, { 
            id: (Date.now() + 1).toString(), 
            role: 'assistant', 
-            content: `Sorry, something went wrong: ${error.message}` 
+            content: `Sorry, something went wrong: ${errorMessage}` 
        }]);
    } finally {
        if (abortControllersRef.current[targetSessionKey] === controller) {
@@ -1383,6 +1723,61 @@ export function ChatInterface() {
            </div>
          ) : (
            <div className="max-w-3xl mx-auto px-4 py-8 space-y-8">
+              {a2aEnabled ? (
+                <div className="rounded-xl border border-border bg-background p-3 space-y-2">
+                  <div className="flex items-center justify-between gap-2">
+                    <div className="text-xs font-semibold text-muted-foreground uppercase tracking-wider">A2A Tasks</div>
+                    <div className="flex items-center gap-2">
+                      <select
+                        value={a2aTaskStateFilter}
+                        onChange={(e) => setA2aTaskStateFilter(e.target.value)}
+                        className="h-7 rounded-md border border-border bg-background px-2 text-xs text-foreground"
+                      >
+                        <option value="all">{t('allStates')}</option>
+                        <option value="SUBMITTED">SUBMITTED</option>
+                        <option value="WORKING">WORKING</option>
+                        <option value="COMPLETED">COMPLETED</option>
+                        <option value="FAILED">FAILED</option>
+                        <option value="CANCELED">CANCELED</option>
+                      </select>
+                      <button
+                        type="button"
+                        className="h-7 px-2 rounded-md border border-border text-xs text-muted-foreground hover:text-foreground"
+                        onClick={() => {
+                          if (currentProject) {
+                            void fetchA2aAgentsAndTasks(currentProject.id, a2aTaskStateFilter);
+                          }
+                        }}
+                      >
+                        {isA2aTaskLoading ? t('loading') : t('refresh')}
+                      </button>
+                    </div>
+                  </div>
+                  <div className="space-y-1.5 max-h-[160px] overflow-y-auto">
+                    {a2aTasks.length === 0 ? (
+                      <div className="text-xs text-muted-foreground">{t('noA2aTasks')}</div>
+                    ) : (
+                      a2aTasks.slice(0, 8).map((task) => (
+                        <div key={task.id} className="flex items-center justify-between gap-2 rounded-md border border-border px-2.5 py-1.5">
+                          <div className="min-w-0">
+                            <div className="text-[11px] text-foreground font-mono truncate">{task.id}</div>
+                            <div className="text-[11px] text-muted-foreground truncate">{task.source} · {task.state}</div>
+                          </div>
+                          {!isTerminalA2aState(task.state) ? (
+                            <button
+                              type="button"
+                              className="text-[11px] px-2 py-1 rounded border border-border text-muted-foreground hover:text-foreground"
+                              onClick={() => void handleCancelA2aTask(task.id)}
+                            >
+                              {t('cancel')}
+                            </button>
+                          ) : null}
+                        </div>
+                      ))
+                    )}
+                  </div>
+                </div>
+              ) : null}
              {messages.map((msg, msgIdx) => {
                const isMessageGenerating = isLoading && msgIdx === messages.length - 1;
                const { markdown, reportHtml } = splitReportHtml(msg.content);
@@ -1414,6 +1809,41 @@ export function ChatInterface() {
                  >
                    {msg.role === "assistant" ? (
                      <>
+                        {msg.a2aTaskId ? (
+                          <div className="mb-3 rounded-xl border border-border bg-muted/50/60 px-3 py-2">
+                            <div className="flex items-center justify-between gap-2">
+                              <div className="text-[11px] text-muted-foreground">
+                                <span className="font-mono">{msg.a2aTaskId}</span>
+                                <span className="mx-1">·</span>
+                                <span>{msg.a2aTaskState || 'SUBMITTED'}</span>
+                              </div>
+                              <div className="flex items-center gap-1">
+                                {msg.a2aTaskState && !isTerminalA2aState(msg.a2aTaskState) ? (
+                                  <button
+                                    type="button"
+                                    className="inline-flex items-center gap-1 rounded-md border border-border px-2 py-0.5 text-[11px] text-muted-foreground hover:text-foreground"
+                                    onClick={() => void handleCancelA2aTask(msg.a2aTaskId!)}
+                                  >
+                                    {t('cancel')}
+                                  </button>
+                                ) : null}
+                                {msg.a2aInputText ? (
+                                  <button
+                                    type="button"
+                                    className="inline-flex items-center gap-1 rounded-md border border-border px-2 py-0.5 text-[11px] text-muted-foreground hover:text-foreground"
+                                    onClick={() => void handleRetryA2aTask(msg)}
+                                  >
+                                    <RotateCcw className="h-3 w-3" />
+                                    {t('retry')}
+                                  </button>
+                                ) : null}
+                              </div>
+                            </div>
+                            {msg.a2aError ? (
+                              <div className="mt-1 text-[11px] text-rose-600">{msg.a2aError}</div>
+                            ) : null}
+                          </div>
+                        ) : null}
                        {displayedThinkingContent && (
                          <div className="mb-3 rounded-xl border border-border bg-muted/50/50 p-3 text-sm text-muted-foreground font-mono whitespace-pre-wrap leading-relaxed shadow-inner">
                            <button
@@ -310,6 +310,36 @@
  "dashScope": "DashScope",
  "volcengine": "Volcengine",
  "tableRowColDesc": "TABLE · {{rowCount}} rows · {{colCount}} columns",
+  "retry": "Retry",
+  "allStates": "All States",
+  "refreshHealth": "Refresh Health",
+  "a2aConfig": "A2A Config",
+  "a2aAgentManagement": "A2A Agent Management",
+  "a2aTaskObservability": "A2A Task Observability",
+  "a2aStatus": "A2A Status",
+  "a2aTaskCreated": "A2A Task Created",
+  "a2aTaskFailed": "A2A Task Failed",
+  "a2aModeEnabled": "A2A Mode: On",
+  "a2aModeDisabled": "A2A Mode: Off",
+  "autoSelectAgent": "Auto Select Agent",
+  "addA2aAgent": "Add A2A Agent",
+  "editA2aAgent": "Edit A2A Agent",
+  "saveA2aAgent": "Save A2A Agent",
+  "a2aAgentName": "A2A Agent Name",
+  "authScheme": "Auth Scheme",
+  "authToken": "Auth Token",
+  "leaveEmptyToKeepUnchanged": "Leave empty to keep unchanged",
+  "healthStatus": "Health",
+  "healthy": "Healthy",
+  "unhealthy": "Unhealthy",
+  "protocol": "Protocol",
+  "capabilities": "Capabilities",
+  "taskId": "Task ID",
+  "taskSource": "Task Source",
+  "time": "Time",
+  "noA2aAgents": "No A2A agents configured",
+  "noA2aTasks": "No A2A tasks",
+  "confirmDeleteA2aAgent": "Are you sure you want to delete this A2A agent?",
  "projectName": "Project Name",
  "dashboardMenu": "Dashboard",
  "newThread": "New Thread",
@@ -325,6 +325,36 @@
  "dashScope": "DashScope (通义千问)",
  "volcengine": "Volcengine (火山引擎)",
  "tableRowColDesc": "TABLE · {{rowCount}} 行 · {{colCount}} 列",
+  "retry": "重试",
+  "allStates": "全部状态",
+  "refreshHealth": "刷新健康检查",
+  "a2aConfig": "A2A 配置",
+  "a2aAgentManagement": "A2A Agent 管理",
+  "a2aTaskObservability": "A2A 任务观测",
+  "a2aStatus": "A2A 状态",
+  "a2aTaskCreated": "A2A 任务已创建",
+  "a2aTaskFailed": "A2A 任务失败",
+  "a2aModeEnabled": "A2A 模式：开启",
+  "a2aModeDisabled": "A2A 模式：关闭",
+  "autoSelectAgent": "自动选择 Agent",
+  "addA2aAgent": "添加 A2A Agent",
+  "editA2aAgent": "编辑 A2A Agent",
+  "saveA2aAgent": "保存 A2A Agent",
+  "a2aAgentName": "A2A Agent 名称",
+  "authScheme": "认证方式",
+  "authToken": "认证 Token",
+  "leaveEmptyToKeepUnchanged": "留空表示不修改",
+  "healthStatus": "健康状态",
+  "healthy": "健康",
+  "unhealthy": "异常",
+  "protocol": "协议版本",
+  "capabilities": "能力",
+  "taskId": "任务 ID",
+  "taskSource": "任务来源",
+  "time": "时间",
+  "noA2aAgents": "暂无 A2A Agent",
+  "noA2aTasks": "暂无 A2A 任务",
+  "confirmDeleteA2aAgent": "确定要删除这个 A2A Agent 吗？",
  "projectName": "项目名称",
  "dashboardMenu": "仪表盘",
  "newThread": "新会话",
@@ -2,13 +2,14 @@ import { useState, useEffect } from 'react';
 import { useTranslation } from 'react-i18next';
 import { Button } from "@/components/ui/button";
 import { Input } from "@/components/ui/input";
-import { Trash2, Terminal, Loader2, FolderOpen, Eye, ShieldCheck, AlertCircle, Wand2, Upload, Plus, RefreshCw } from "lucide-react";
+import { Trash2, Terminal, Loader2, FolderOpen, Eye, ShieldCheck, AlertCircle, Wand2, Upload, Plus, RefreshCw, HeartPulse } from "lucide-react";
 import { Dialog, DialogContent, DialogHeader, DialogTitle, DialogFooter } from "@/components/ui/dialog";
 import { Label } from "@/components/ui/label";
 import { Textarea } from "@/components/ui/textarea";
 import { Select, SelectContent, SelectItem, SelectTrigger, SelectValue } from "@/components/ui/select";
 import { Table, TableBody, TableCell, TableHead, TableHeader, TableRow } from "@/components/ui/table";
 import { api } from "@/lib/api";
+import { a2aApi, type A2ARemoteAgent, type A2ATask } from "@/api/a2a";
 import { useProjectStore } from "@/store/projectStore";
 import { useMcpHealthStore } from "@/store/mcpHealthStore";
 import { useRef } from 'react';
@@ -40,6 +41,13 @@ interface MCPServer {
  status?: string;
 }

+interface A2ARemoteAgentForm {
+  name: string;
+  base_url: string;
+  auth_scheme: "none" | "bearer";
+  auth_token: string;
+}
+
 const SOURCE_LOCAL_IMPORT = "local_import";
 const SOURCE_SYSTEM_BUILTIN = "system_builtin";
 const SOURCE_BACKEND_GENERATED = "backend_generated";
@@ -76,7 +84,7 @@ const dedupeSkillsById = (skills: Skill[]): Skill[] => {

 export function Skills() {
  const { t } = useTranslation();
-  const [activeTab, setActiveTab] = useState<'skills' | 'mcp'>('skills');
+  const [activeTab, setActiveTab] = useState<'skills' | 'mcp' | 'a2a'>('skills');
  const [sourceFilter, setSourceFilter] = useState<string>('all');

  // Skills state
@@ -97,6 +105,20 @@ export function Skills() {
  const [mcpHeadersStr, setMcpHeadersStr] = useState('');
  const [isRefreshingMcpHealth, setIsRefreshingMcpHealth] = useState(false);

+  const [a2aAgents, setA2aAgents] = useState<A2ARemoteAgent[]>([]);
+  const [a2aTasks, setA2aTasks] = useState<A2ATask[]>([]);
+  const [a2aTaskStateFilter, setA2aTaskStateFilter] = useState<string>('all');
+  const [isA2aLoading, setIsA2aLoading] = useState(false);
+  const [isA2aDialogOpen, setIsA2aDialogOpen] = useState(false);
+  const [editingA2aAgent, setEditingA2aAgent] = useState<A2ARemoteAgent | null>(null);
+  const [a2aForm, setA2aForm] = useState<A2ARemoteAgentForm>({
+    name: '',
+    base_url: '',
+    auth_scheme: 'none',
+    auth_token: '',
+  });
+  const [isA2aRefreshingHealth, setIsA2aRefreshingHealth] = useState(false);
+
  const { currentProject } = useProjectStore();
  const { hasMcpError, refresh: refreshMcpHealth } = useMcpHealthStore();
  const fileInputRef = useRef<HTMLInputElement>(null);
@@ -136,15 +158,34 @@ export function Skills() {
      }
    };

+    const fetchA2aData = async () => {
+      if (!currentProject) return;
+      setIsA2aLoading(true);
+      try {
+        const [agents, tasks] = await Promise.all([
+          a2aApi.listRemoteAgents(currentProject.id),
+          a2aApi.listTasks(currentProject.id, a2aTaskStateFilter),
+        ]);
+        setA2aAgents(agents || []);
+        setA2aTasks(tasks || []);
+      } catch (error) {
+        console.error("Failed to fetch A2A data", error);
+      } finally {
+        setIsA2aLoading(false);
+      }
+    };
+
    if (currentProject) {
      void refreshMcpHealth(currentProject.id);
      if (activeTab === 'skills') {
        void fetchSkills();
-      } else {
+      } else if (activeTab === 'mcp') {
        void fetchMcpServers();
+      } else {
+        void fetchA2aData();
      }
    }
-  }, [currentProject?.id, activeTab, refreshMcpHealth]);
+  }, [currentProject, currentProject?.id, activeTab, refreshMcpHealth, a2aTaskStateFilter]);

  const fetchSkills = async () => {
    if (!currentProject) return;
@@ -194,6 +235,100 @@ export function Skills() {
    }
  };

+  const fetchA2aData = async () => {
+    if (!currentProject) return;
+    setIsA2aLoading(true);
+    try {
+      const [agents, tasks] = await Promise.all([
+        a2aApi.listRemoteAgents(currentProject.id),
+        a2aApi.listTasks(currentProject.id, a2aTaskStateFilter),
+      ]);
+      setA2aAgents(agents || []);
+      setA2aTasks(tasks || []);
+    } catch (error) {
+      console.error("Failed to fetch A2A data", error);
+    } finally {
+      setIsA2aLoading(false);
+    }
+  };
+
+  const handleRefreshA2aHealth = async () => {
+    if (!currentProject || a2aAgents.length === 0) return;
+    setIsA2aRefreshingHealth(true);
+    try {
+      await Promise.all(a2aAgents.map((agent) => a2aApi.healthCheckRemoteAgent(agent.id)));
+      await fetchA2aData();
+    } finally {
+      setIsA2aRefreshingHealth(false);
+    }
+  };
+
+  const handleOpenCreateA2a = () => {
+    setEditingA2aAgent(null);
+    setA2aForm({
+      name: '',
+      base_url: '',
+      auth_scheme: 'none',
+      auth_token: '',
+    });
+    setIsA2aDialogOpen(true);
+  };
+
+  const handleOpenEditA2a = (agent: A2ARemoteAgent) => {
+    setEditingA2aAgent(agent);
+    setA2aForm({
+      name: agent.name,
+      base_url: agent.base_url,
+      auth_scheme: agent.auth_scheme,
+      auth_token: '',
+    });
+    setIsA2aDialogOpen(true);
+  };
+
+  const handleSaveA2aAgent = async () => {
+    if (!currentProject) return;
+    if (!a2aForm.name.trim() || !a2aForm.base_url.trim()) return;
+    const payload = {
+      name: a2aForm.name.trim(),
+      base_url: a2aForm.base_url.trim(),
+      auth_scheme: a2aForm.auth_scheme,
+      ...(a2aForm.auth_scheme === 'bearer' && a2aForm.auth_token.trim() ? { auth_token: a2aForm.auth_token.trim() } : {}),
+    };
+    try {
+      if (editingA2aAgent) {
+        await a2aApi.updateRemoteAgent(editingA2aAgent.id, payload);
+      } else {
+        await a2aApi.createRemoteAgent({
+          project_id: currentProject.id,
+          ...payload,
+        });
+      }
+      setIsA2aDialogOpen(false);
+      await fetchA2aData();
+    } catch (error) {
+      console.error("Failed to save A2A agent", error);
+    }
+  };
+
+  const handleDeleteA2aAgent = async (agentId: number) => {
+    if (!window.confirm(t('confirmDeleteA2aAgent'))) return;
+    try {
+      await a2aApi.deleteRemoteAgent(agentId);
+      await fetchA2aData();
+    } catch (error) {
+      console.error("Failed to delete A2A agent", error);
+    }
+  };
+
+  const handleRefreshA2aCard = async (agentId: number) => {
+    try {
+      await a2aApi.refreshRemoteAgentCard(agentId);
+      await fetchA2aData();
+    } catch (error) {
+      console.error("Failed to refresh A2A card", error);
+    }
+  };
+
  const handleFileUpload = async (event: React.ChangeEvent<HTMLInputElement>) => {
    const file = event.target.files?.[0];
    if (!file || !currentProject) return;
@@ -371,6 +506,12 @@ export function Skills() {
                <span className="absolute top-1 right-1 w-1.5 h-1.5 rounded-full bg-red-500 animate-pulse" title="MCP Server Error" />
              )}
            </button>
+            <button
+              className={`px-3 py-1 text-sm font-medium rounded-md transition-colors ${activeTab === 'a2a' ? 'bg-background shadow-sm text-foreground' : 'text-muted-foreground hover:text-foreground/80'}`}
+              onClick={() => setActiveTab('a2a')}
+            >
+              {t('a2aConfig')}
+            </button>
          </div>
          {activeTab === 'skills' ? (
            <>
@@ -407,7 +548,7 @@ export function Skills() {
                {isLoading ? t('uploading', '上传中...') : t('uploadSkill')}
              </Button>
            </>
-          ) : (
+          ) : activeTab === 'mcp' ? (
            <>
              <Button
                variant="outline"
@@ -422,13 +563,52 @@ export function Skills() {
                )}
                {t('refresh')}
              </Button>
-              <Button 
+              <Button
                className="h-9 bg-[#ff4d29] hover:bg-[#ff4d29]/90 text-white gap-2 rounded-md px-3"
                onClick={() => setIsMcpDialogOpen(true)}
              >
                <Plus className="h-4 w-4" />{t('addMcpServer')}
              </Button>
            </>
+          ) : (
+            <>
+              <Select value={a2aTaskStateFilter} onValueChange={(val) => { if (val) setA2aTaskStateFilter(val); }}>
+                <SelectTrigger className="w-[150px] h-9">
+                  <SelectValue />
+                </SelectTrigger>
+                <SelectContent>
+                  <SelectItem value="all">{t('allStates')}</SelectItem>
+                  <SelectItem value="SUBMITTED">SUBMITTED</SelectItem>
+                  <SelectItem value="WORKING">WORKING</SelectItem>
+                  <SelectItem value="COMPLETED">COMPLETED</SelectItem>
+                  <SelectItem value="FAILED">FAILED</SelectItem>
+                  <SelectItem value="CANCELED">CANCELED</SelectItem>
+                </SelectContent>
+              </Select>
+              <Button
+                variant="outline"
+                className="h-9 gap-2 rounded-md px-3"
+                onClick={handleRefreshA2aHealth}
+                disabled={isA2aRefreshingHealth || a2aAgents.length === 0}
+              >
+                {isA2aRefreshingHealth ? <Loader2 className="h-4 w-4 animate-spin" /> : <HeartPulse className="h-4 w-4" />}
+                {t('refreshHealth')}
+              </Button>
+              <Button
+                variant="outline"
+                className="h-9 gap-2 rounded-md px-3"
+                onClick={() => void fetchA2aData()}
+              >
+                <RefreshCw className="h-4 w-4" />
+                {t('refresh')}
+              </Button>
+              <Button
+                className="h-9 bg-[#ff4d29] hover:bg-[#ff4d29]/90 text-white gap-2 rounded-md px-3"
+                onClick={handleOpenCreateA2a}
+              >
+                <Plus className="h-4 w-4" />{t('addA2aAgent')}
+              </Button>
+            </>
          )}
        </div>
      </div>
@@ -555,7 +735,7 @@ export function Skills() {
            </TableBody>
          </Table>
        </div>
-        ) : (
+        ) : activeTab === 'mcp' ? (
          <div className="bg-background rounded-xl border border-border shadow-sm overflow-hidden min-w-[800px] lg:min-w-0">
            <Table className="table-fixed w-full">
              <TableHeader className="bg-muted/50/50">
@@ -646,6 +826,112 @@ export function Skills() {
              </TableBody>
            </Table>
          </div>
+        ) : (
+          <div className="space-y-4">
+            <div className="bg-background rounded-xl border border-border shadow-sm overflow-hidden min-w-[800px] lg:min-w-0">
+              <div className="px-4 py-3 border-b border-border text-sm font-semibold text-foreground/80">
+                {t('a2aAgentManagement')}
+              </div>
+              <Table className="table-fixed w-full">
+                <TableHeader className="bg-muted/50/50">
+                  <TableRow className="hover:bg-transparent">
+                    <TableHead className="w-[20%] font-semibold text-foreground/80 py-3 px-4 text-sm">{t('name')}</TableHead>
+                    <TableHead className="w-[24%] font-semibold text-foreground/80 py-3 px-4 text-sm">{t('url')}</TableHead>
+                    <TableHead className="w-[10%] font-semibold text-foreground/80 py-3 px-4 text-sm">{t('protocol')}</TableHead>
+                    <TableHead className="w-[16%] font-semibold text-foreground/80 py-3 px-4 text-sm">{t('capabilities')}</TableHead>
+                    <TableHead className="w-[15%] font-semibold text-foreground/80 py-3 px-4 text-sm">{t('healthStatus')}</TableHead>
+                    <TableHead className="w-[15%] font-semibold text-foreground/80 py-3 px-4 text-sm text-right">{t('actions')}</TableHead>
+                  </TableRow>
+                </TableHeader>
+                <TableBody>
+                  {isA2aLoading ? (
+                    <TableRow>
+                      <TableCell colSpan={6} className="py-20 text-center">
+                        <Loader2 className="h-8 w-8 animate-spin text-indigo-500 mx-auto" />
+                      </TableCell>
+                    </TableRow>
+                  ) : a2aAgents.length === 0 ? (
+                    <TableRow>
+                      <TableCell colSpan={6} className="py-16 text-center text-muted-foreground">{t('noA2aAgents')}</TableCell>
+                    </TableRow>
+                  ) : (
+                    a2aAgents.map((agent) => (
+                      <TableRow key={agent.id} className="group hover:bg-muted/50/50 transition-colors border-border">
+                        <TableCell className="py-4 px-4 text-sm font-medium">{agent.name}</TableCell>
+                        <TableCell className="py-4 px-4 text-sm text-muted-foreground truncate" title={agent.base_url}>{agent.base_url}</TableCell>
+                        <TableCell className="py-4 px-4 text-sm text-muted-foreground">{agent.protocol_version || '-'}</TableCell>
+                        <TableCell className="py-4 px-4 text-sm text-muted-foreground truncate" title={agent.capabilities.join(', ')}>{agent.capabilities.join(', ') || '-'}</TableCell>
+                        <TableCell className="py-4 px-4">
+                          <div className={`inline-flex items-center gap-1 px-2 py-0.5 rounded-full text-[10px] md:text-xs font-medium whitespace-nowrap ${agent.healthy ? 'bg-green-50 text-green-700 border border-green-100' : 'bg-rose-50 text-rose-700 border border-rose-100'}`}>
+                            {agent.healthy ? <ShieldCheck className="h-3 w-3" /> : <AlertCircle className="h-3 w-3" />}
+                            {agent.healthy ? t('healthy') : t('unhealthy')}
+                            <span className="opacity-70">#{agent.failure_count}</span>
+                          </div>
+                        </TableCell>
+                        <TableCell className="py-4 px-4 text-right">
+                          <div className="flex items-center justify-end gap-1">
+                            <Button variant="ghost" size="icon" className="h-8 w-8 text-muted-foreground hover:text-indigo-600 hover:bg-indigo-50 rounded-md transition-all shrink-0" onClick={() => void handleRefreshA2aCard(agent.id)}>
+                              <RefreshCw className="h-4 w-4" />
+                            </Button>
+                            <Button variant="ghost" size="icon" className="h-8 w-8 text-muted-foreground hover:text-indigo-600 hover:bg-indigo-50 rounded-md transition-all shrink-0" onClick={() => handleOpenEditA2a(agent)}>
+                              <Eye className="h-4 w-4" />
+                            </Button>
+                            <Button variant="ghost" size="icon" className="h-8 w-8 text-muted-foreground hover:text-rose-600 hover:bg-rose-50 rounded-md transition-all shrink-0" onClick={() => void handleDeleteA2aAgent(agent.id)}>
+                              <Trash2 className="h-4 w-4" />
+                            </Button>
+                          </div>
+                        </TableCell>
+                      </TableRow>
+                    ))
+                  )}
+                </TableBody>
+              </Table>
+            </div>
+
+            <div className="bg-background rounded-xl border border-border shadow-sm overflow-hidden min-w-[800px] lg:min-w-0">
+              <div className="px-4 py-3 border-b border-border text-sm font-semibold text-foreground/80">
+                {t('a2aTaskObservability')}
+              </div>
+              <Table className="table-fixed w-full">
+                <TableHeader className="bg-muted/50/50">
+                  <TableRow className="hover:bg-transparent">
+                    <TableHead className="w-[18%] font-semibold text-foreground/80 py-3 px-4 text-sm">{t('taskId')}</TableHead>
+                    <TableHead className="w-[12%] font-semibold text-foreground/80 py-3 px-4 text-sm">{t('taskSource')}</TableHead>
+                    <TableHead className="w-[12%] font-semibold text-foreground/80 py-3 px-4 text-sm">{t('status')}</TableHead>
+                    <TableHead className="w-[38%] font-semibold text-foreground/80 py-3 px-4 text-sm">{t('content')}</TableHead>
+                    <TableHead className="w-[20%] font-semibold text-foreground/80 py-3 px-4 text-sm">{t('time')}</TableHead>
+                  </TableRow>
+                </TableHeader>
+                <TableBody>
+                  {isA2aLoading ? (
+                    <TableRow>
+                      <TableCell colSpan={5} className="py-16 text-center">
+                        <Loader2 className="h-8 w-8 animate-spin text-indigo-500 mx-auto" />
+                      </TableCell>
+                    </TableRow>
+                  ) : a2aTasks.length === 0 ? (
+                    <TableRow>
+                      <TableCell colSpan={5} className="py-14 text-center text-muted-foreground">{t('noA2aTasks')}</TableCell>
+                    </TableRow>
+                  ) : (
+                    a2aTasks.map((task) => (
+                      <TableRow key={task.id} className="group hover:bg-muted/50/50 transition-colors border-border">
+                        <TableCell className="py-4 px-4 text-xs font-mono truncate" title={task.id}>{task.id}</TableCell>
+                        <TableCell className="py-4 px-4 text-sm text-muted-foreground">{task.source}</TableCell>
+                        <TableCell className="py-4 px-4 text-sm">{task.state}</TableCell>
+                        <TableCell className="py-4 px-4 text-xs text-muted-foreground">
+                          <div className="line-clamp-2" title={task.error_message || task.output_text || task.input_text}>
+                            {task.error_message || task.output_text || task.input_text}
+                          </div>
+                        </TableCell>
+                        <TableCell className="py-4 px-4 text-xs text-muted-foreground">{task.updated_at}</TableCell>
+                      </TableRow>
+                    ))
+                  )}
+                </TableBody>
+              </Table>
+            </div>
+          </div>
        )}
      </div>

@@ -847,6 +1133,76 @@ export function Skills() {
          </DialogFooter>
        </DialogContent>
      </Dialog>
+
+      <Dialog open={isA2aDialogOpen} onOpenChange={(open) => {
+          setIsA2aDialogOpen(open);
+          if (!open) {
+            setEditingA2aAgent(null);
+            setA2aForm({
+              name: '',
+              base_url: '',
+              auth_scheme: 'none',
+              auth_token: '',
+            });
+          }
+      }}>
+        <DialogContent className="sm:max-w-[600px] max-h-[90vh] flex flex-col rounded-2xl p-0 overflow-hidden">
+          <DialogHeader className="p-6 pb-2">
+            <DialogTitle className="text-xl font-bold text-foreground">{editingA2aAgent ? t('editA2aAgent') : t('addA2aAgent')}</DialogTitle>
+          </DialogHeader>
+          <div className="flex-1 overflow-y-auto px-6 py-2">
+            <div className="grid gap-5">
+              <div className="grid gap-1.5">
+                <Label htmlFor="a2a-name" className="text-muted-foreground font-medium text-sm">{t('name')}</Label>
+                <Input
+                  id="a2a-name"
+                  placeholder={t('a2aAgentName')}
+                  value={a2aForm.name}
+                  onChange={(e) => setA2aForm({ ...a2aForm, name: e.target.value })}
+                  className="rounded-lg border-border h-10"
+                />
+              </div>
+              <div className="grid gap-1.5">
+                <Label htmlFor="a2a-url" className="text-muted-foreground font-medium text-sm">{t('baseUrl')}</Label>
+                <Input
+                  id="a2a-url"
+                  placeholder="https://example-agent.com"
+                  value={a2aForm.base_url}
+                  onChange={(e) => setA2aForm({ ...a2aForm, base_url: e.target.value })}
+                  className="rounded-lg border-border h-10"
+                />
+              </div>
+              <div className="grid gap-1.5">
+                <Label htmlFor="a2a-auth-scheme" className="text-muted-foreground font-medium text-sm">{t('authScheme')}</Label>
+                <Select value={a2aForm.auth_scheme} onValueChange={(val) => { if (val) setA2aForm({ ...a2aForm, auth_scheme: val as "none" | "bearer" }); }}>
+                  <SelectTrigger className="rounded-lg border-border h-10">
+                    <SelectValue />
+                  </SelectTrigger>
+                  <SelectContent className="rounded-lg">
+                    <SelectItem value="none">none</SelectItem>
+                    <SelectItem value="bearer">bearer</SelectItem>
+                  </SelectContent>
+                </Select>
+              </div>
+              {a2aForm.auth_scheme === 'bearer' ? (
+                <div className="grid gap-1.5">
+                  <Label htmlFor="a2a-auth-token" className="text-muted-foreground font-medium text-sm">{t('authToken')}</Label>
+                  <Input
+                    id="a2a-auth-token"
+                    placeholder={editingA2aAgent ? t('leaveEmptyToKeepUnchanged') : t('enterApiKey')}
+                    value={a2aForm.auth_token}
+                    onChange={(e) => setA2aForm({ ...a2aForm, auth_token: e.target.value })}
+                    className="rounded-lg border-border h-10"
+                  />
+                </div>
+              ) : null}
+            </div>
+          </div>
+          <DialogFooter className="p-6 pt-2">
+            <Button onClick={handleSaveA2aAgent} className="bg-indigo-600 hover:bg-indigo-700 text-primary-foreground rounded-lg px-6 h-10 w-full">{t('saveA2aAgent')}</Button>
+          </DialogFooter>
+        </DialogContent>
+      </Dialog>
    </div>
  );
 }