Un serviciu generează inventarul aprobat pentru trei componente software, iar alt serviciu trebuie să confirme că rezultatul primit prin rețea nu a fost modificat. Vom transforma această cerință într-un flux local cu trei procese Python: un orchestrator Google ADK, un server MCP și un agent ADK expus prin protocolul A2A.
La final, comanda de test va trimite aceeași cerere prin întregul traseu, va recalcula local digestul SHA-256 și va accepta rezultatul numai dacă manifestul MCP, răspunsul agentului remote și fișierul salvat coincid. Proiectul nu folosește un model generativ și nu cere cheie API: agenții sunt implementați determinist prin BaseAgent, astfel încât verificarea să nu depindă de formularea imprevizibilă a unui LLM.
Trei servicii locale și o condiție strictă de acceptare
Scenariul este cel al unei echipe care pregătește un inventar pentru o lansare. Orchestratorul primește lista componentelor, iar serviciul MCP o normalizează, o sortează și produce un manifest semnat logic prin SHA-256. Agentul remote primește manifestul prin A2A, recalculează digestul și răspunde cu verified: true numai dacă datele sunt intacte.
Fluxul este:
orchestrator/run.pypornește un agent ADK local.- Agentul deschide o sesiune Streamable HTTP către
http://127.0.0.1:8000/mcp. - Unealta MCP
build_manifestproduce manifestul și digestul. - Orchestratorul citește Agent Card-ul de la
http://127.0.0.1:8001/.well-known/agent-card.json. - Manifestul este trimis agentului remote prin metoda A2A
message/send. - Răspunsul remote este comparat cu rezultatul MCP.
- Numai după toate verificările apare
runs/result.json.
Acest proiect verifică integritatea schimbului, nu identitatea serviciilor. Conexiunile sunt limitate la interfața loopback, potrivită laboratorului local. Pentru o rețea reală ar mai fi necesare TLS, autentificare, autorizare, rotația credențialelor și o politică explicită de încredere pentru Agent Card.
Separarea planificării de execuție rămâne relevantă chiar dacă aici nu folosim un LLM. Articolul despre controlul dintre planificare și execuție în agenți AI explică de ce rezultatul unei unelte nu ar trebui transformat automat într-o acțiune acceptată.
Fixăm laboratorul la ADK 2.10.0 și MCP 2.0.0
Tutorialul folosește Ubuntu Server 24.04.5 LTS, Python 3.12 și următoarele versiuni fixate:
google-adk[a2a]==2.10.0;mcp==2.0.0;httpx==0.28.1;uvicorn==0.37.0;pytest==8.4.2.
Sunt suficiente două nuclee CPU, 2 GB RAM și aproximativ 1 GB liber. Nu este necesar un GPU. Configurarea și testul durează aproximativ 45–60 de minute.
Verifică gazda înainte de instalare:
cat /etc/os-release
python3 --version
uname -m
free -h
df -h .Un exemplu de rezultat compatibil este:
PRETTY_NAME="Ubuntu 24.04.5 LTS"
Python 3.12.x
x86_64
Memorie disponibilă: cel puțin 2 GiB
Spațiu disponibil în directorul curent: cel puțin 1 GiBNumărul exact al patch-ului Python, memoria liberă și valorile discului pot fi diferite. Oprește tutorialul dacă python3 --version indică o versiune mai veche decât 3.12 sau dacă sistemul nu este Ubuntu 24.04 LTS; fișierul de dependențe de mai jos a fost ales pentru mediul declarat, nu pentru o combinație arbitrară.
Instalarea comună rămâne într-un singur bloc:
sudo apt update
sudo apt install -y python3-venv curl jq
mkdir -p "$HOME/labs/adk-a2a-mcp"
cd "$HOME/labs/adk-a2a-mcp"
python3 -m venv .venv
source .venv/bin/activate
python -m pip install --upgrade pipPachetele sunt instalate numai în mediul virtual. Niciunul dintre cele trei servicii nu rulează cu sudo.
Fișierele delimitează protocolul, logica și dovada
Arborele complet va arăta astfel:
adk-a2a-mcp/
├── data/
│ └── request.json # cererea fixă trimisă orchestratorului
├── logs/ # stdout și stderr pentru cele două servicii
├── orchestrator/
│ ├── __init__.py
│ ├── agent.py # agentul ADK local și clienții MCP/A2A
│ └── run.py # sesiunea ADK și salvarea rezultatului
├── remote_agent/
│ ├── __init__.py
│ ├── agent.py # recalcularea digestului primit
│ └── server.py # expunerea agentului prin to_a2a()
├── runs/ # rezultatul acceptat
├── scripts/
│ ├── start.sh # pornirea și verificarea porturilor locale
│ ├── stop.sh # oprirea proceselor lansate de proiect
│ └── smoke.sh # testul reproductibil de capăt
├── services/
│ ├── __init__.py
│ └── mcp_service.py # unealta MCP build_manifest
├── tests/
│ └── test_result.py # verificarea independentă a fișierului
└── requirements.txtDirectorul logs păstrează ieșirea proceselor, iar runs conține numai rezultate care au trecut verificarea locală. Fișierele PID sunt create în runs, nu într-un director global.
Pasul 1 — Înghețăm dependențele și cererea de inventar
De ce: versiunile fixate împiedică instalarea unei ediții ulterioare cu API diferit. Cererea fixă oferă aceeași intrare fiecărei rulări și permite compararea rezultatului.
Creează directoarele și fișierele de pachet:
cd "$HOME/labs/adk-a2a-mcp"
mkdir -p data logs orchestrator remote_agent runs scripts services tests
touch orchestrator/__init__.py remote_agent/__init__.py services/__init__.pyFișierul requirements.txt:
google-adk[a2a]==2.10.0
httpx==0.28.1
mcp==2.0.0
pytest==8.4.2
uvicorn==0.37.0Fișierul data/request.json:
{
"items": [
{
"name": "api-comenzi",
"version": "2026.10.0"
},
{
"name": "worker-facturi",
"version": "2026.10.0"
},
{
"name": "portal-clienti",
"version": "2026.10.0"
}
]
}Instalează exact dependențele declarate și verifică rezolvarea:
source .venv/bin/activate
python -m pip install -r requirements.txt
python -m pip check
python -m pip show google-adk mcp httpx uvicorn pytestpip check trebuie să afișeze:
No broken requirements found.Opțiunea [a2a] instalează componentele necesare integrării A2A din ADK. httpx este folosit explicit de orchestrator, iar uvicorn servește aplicația ASGI produsă de to_a2a().
Punct de control: dacă pip check raportează un conflict, confirmă că promptul începe cu (.venv) și recreează mediul virtual; nu elimina constrângerile de versiune pentru a forța instalarea.
Pasul 2 — Producem manifestul printr-o unealtă MCP izolată
De ce: MCP delimitează accesul la funcția externă. Orchestratorul nu reproduce logica de construire a manifestului, ci primește un rezultat serializat de serviciul de pe portul 8000.
Fișierul services/mcp_service.py:
from __future__ import annotations
import hashlib
import json
from typing import Any
from mcp.server.fastmcp import FastMCP
mcp = FastMCP(
"manifest-service",
host="127.0.0.1",
port=8000,
stateless_http=True,
)
def canonical_json(value: Any) -> str:
return json.dumps(
value,
ensure_ascii=False,
separators=(",", ":"),
sort_keys=True,
)
@mcp.tool()
def build_manifest(items: list[dict[str, str]]) -> str:
"""Normalizează componentele și atașează digestul SHA-256."""
if not items:
raise ValueError("items must contain at least one component")
normalized: list[dict[str, str]] = []
seen: set[str] = set()
for index, item in enumerate(items):
name = item.get("name", "").strip()
version = item.get("version", "").strip()
if not name or not version:
raise ValueError(
f"item {index} must contain non-empty name and version"
)
if name in seen:
raise ValueError(f"duplicate component name: {name}")
seen.add(name)
normalized.append({"name": name, "version": version})
normalized.sort(key=lambda component: component["name"])
manifest = {
"count": len(normalized),
"items": normalized,
"schema": "cyberai.inventory.v1",
}
digest = hashlib.sha256(
canonical_json(manifest).encode("utf-8")
).hexdigest()
return canonical_json(
{
"manifest": manifest,
"sha256": digest,
}
)
if __name__ == "__main__":
mcp.run(transport="streamable-http")sort_keys=True și separatorii fără spații fac serializarea canonică: aceeași structură produce aceiași octeți. Sortarea componentelor după name elimină influența ordinii din cerere. stateless_http=True este potrivit uneltei care nu păstrează stare între apeluri.
Digestul nu este o semnătură criptografică și nu dovedește cine a creat manifestul. El detectează diferențele numai pentru că orchestratorul și agentul remote recalculează aceeași reprezentare.
Pornește temporar serviciul:
source .venv/bin/activate
python -m services.mcp_serviceÎn alt terminal, verifică faptul că portul este deschis numai local:
ss -ltn | grep '127.0.0.1:8000'Exemplu de rezultat:
LISTEN 0 128 127.0.0.1:8000 0.0.0.0:*Punct de control: dacă vezi 0.0.0.0:8000, oprește serviciul și verifică argumentul host; laboratorul nu trebuie expus altor sisteme.
Pasul 3 — Expunem verificatorul ADK ca agent A2A
De ce: agentul remote reprezintă o graniță separată de serviciu. El nu are acces la starea internă a orchestratorului și acceptă numai textul primit în mesajul A2A.
Fișierul remote_agent/agent.py:
from __future__ import annotations
import hashlib
import json
from collections.abc import AsyncGenerator
from typing import Any
from google.adk.agents import BaseAgent
from google.adk.agents.invocation_context import InvocationContext
from google.adk.events import Event
from google.genai import types
def canonical_json(value: Any) -> str:
return json.dumps(
value,
ensure_ascii=False,
separators=(",", ":"),
sort_keys=True,
)
class DigestVerifierAgent(BaseAgent):
async def _run_async_impl(
self,
ctx: InvocationContext,
) -> AsyncGenerator[Event, None]:
parts = ctx.user_content.parts if ctx.user_content else []
raw_text = "".join(part.text or "" for part in parts)
try:
envelope = json.loads(raw_text)
manifest = envelope["manifest"]
claimed_digest = envelope["sha256"]
actual_digest = hashlib.sha256(
canonical_json(manifest).encode("utf-8")
).hexdigest()
items = manifest.get("items")
declared_count = manifest.get("count")
schema_ok = manifest.get("schema") == "cyberai.inventory.v1"
count_ok = (
isinstance(items, list)
and declared_count == len(items)
)
digest_ok = actual_digest == claimed_digest
verified = schema_ok and count_ok and digest_ok
response = {
"count": declared_count,
"received_sha256": claimed_digest,
"recomputed_sha256": actual_digest,
"verified": verified,
}
except (KeyError, TypeError, json.JSONDecodeError) as exc:
response = {
"error": f"invalid manifest envelope: {exc}",
"verified": False,
}
yield Event(
author=self.name,
content=types.Content(
role="model",
parts=[
types.Part(
text=canonical_json(response)
)
],
),
)
root_agent = DigestVerifierAgent(
name="remote_digest_verifier",
description="Verifică schema, numărul elementelor și SHA-256.",
)Fișierul remote_agent/server.py:
from google.adk.a2a.utils.agent_to_a2a import to_a2a
from remote_agent.agent import root_agent
app = to_a2a(
root_agent,
host="127.0.0.1",
port=8001,
)Metoda _run_async_impl() este punctul de extensie pentru un agent ADK personalizat. Agentul produce un Event, nu întoarce direct un șir. to_a2a() transformă agentul într-o aplicație ASGI și generează metadatele de descoperire A2A.
Pornește aplicația:
source .venv/bin/activate
uvicorn remote_agent.server:app
--host 127.0.0.1
--port 8001Din alt terminal, citește Agent Card-ul:
curl --fail --silent
http://127.0.0.1:8001/.well-known/agent-card.json |
jq '{name, url}'Exemplul trebuie să conțină numele agentului și un URL de pe portul 8001:
{
"name": "remote_digest_verifier",
"url": "http://127.0.0.1:8001/"
}Forma cardului poate include și alte câmpuri generate de versiunea ADK fixată. Nu modifica manual cardul în acest laborator.
Punct de control: dacă cererea întoarce codul HTTP 404, verifică logul Uvicorn și confirmă că aplicația importată este remote_agent.server:app, nu obiectul root_agent.
Pasul 4 — Agentul local leagă sesiunea MCP de mesajul A2A
De ce: orchestratorul trebuie să păstreze rezultatul MCP neschimbat, să descopere endpoint-ul A2A și să respingă un răspuns care nu conține dovada așteptată. Nu lăsăm unui model generativ decizia de acceptare.
Fișierul orchestrator/agent.py:
from __future__ import annotations
import json
from collections.abc import AsyncGenerator
from typing import Any
import httpx
from google.adk.agents import BaseAgent
from google.adk.agents.invocation_context import InvocationContext
from google.adk.events import Event
from google.genai import types
from mcp import ClientSession
from mcp.client.streamable_http import streamable_http_client
def canonical_json(value: Any) -> str:
return json.dumps(
value,
ensure_ascii=False,
separators=(",", ":"),
sort_keys=True,
)
def find_verification(value: Any) -> dict[str, Any] | None:
if isinstance(value, dict):
if "verified" in value and "recomputed_sha256" in value:
return value
for child in value.values():
found = find_verification(child)
if found is not None:
return found
if isinstance(value, list):
for child in value:
found = find_verification(child)
if found is not None:
return found
if isinstance(value, str):
try:
decoded = json.loads(value)
except json.JSONDecodeError:
return None
return find_verification(decoded)
return None
class InventoryOrchestrator(BaseAgent):
mcp_url: str
agent_card_url: str
async def call_manifest_tool(
self,
items: list[dict[str, str]],
) -> dict[str, Any]:
async with streamable_http_client(
self.mcp_url
) as (read_stream, write_stream, _):
async with ClientSession(
read_stream,
write_stream,
) as session:
await session.initialize()
result = await session.call_tool(
"build_manifest",
{"items": items},
)
if result.isError:
raise RuntimeError("MCP build_manifest returned an error")
if not result.content:
raise RuntimeError("MCP build_manifest returned no content")
text = getattr(result.content[0], "text", None)
if not text:
raise RuntimeError("MCP result does not contain text")
return json.loads(text)
async def call_remote_agent(
self,
envelope: dict[str, Any],
) -> dict[str, Any]:
timeout = httpx.Timeout(15.0)
async with httpx.AsyncClient(timeout=timeout) as client:
card_response = await client.get(self.agent_card_url)
card_response.raise_for_status()
card = card_response.json()
rpc_url = card.get("url")
if not isinstance(rpc_url, str):
raise RuntimeError("Agent Card does not contain url")
request = {
"id": "inventory-run-001",
"jsonrpc": "2.0",
"method": "message/send",
"params": {
"message": {
"kind": "message",
"messageId": "inventory-message-001",
"parts": [
{
"kind": "text",
"text": canonical_json(envelope),
}
],
"role": "user",
}
},
}
response = await client.post(rpc_url, json=request)
response.raise_for_status()
payload = response.json()
if "error" in payload:
raise RuntimeError(
f"A2A returned an error: {payload['error']}"
)
verification = find_verification(payload.get("result"))
if verification is None:
raise RuntimeError(
"A2A response contains no verification artifact"
)
return verification
async def _run_async_impl(
self,
ctx: InvocationContext,
) -> AsyncGenerator[Event, None]:
parts = ctx.user_content.parts if ctx.user_content else []
request_text = "".join(part.text or "" for part in parts)
request = json.loads(request_text)
envelope = await self.call_manifest_tool(request["items"])
remote = await self.call_remote_agent(envelope)
digest = envelope["sha256"]
accepted = (
remote.get("verified") is True
and remote.get("received_sha256") == digest
and remote.get("recomputed_sha256") == digest
and remote.get("count")
== envelope["manifest"]["count"]
)
if not accepted:
raise RuntimeError(
"Remote verification does not match the MCP result"
)
result = {
"accepted": True,
"manifest": envelope["manifest"],
"remote": remote,
"sha256": digest,
}
yield Event(
author=self.name,
content=types.Content(
role="model",
parts=[
types.Part(
text=canonical_json(result)
)
],
),
)
root_agent = InventoryOrchestrator(
name="inventory_orchestrator",
description="Leagă manifestul MCP de verificarea A2A.",
mcp_url="http://127.0.0.1:8000/mcp",
agent_card_url=(
"http://127.0.0.1:8001/"
".well-known/agent-card.json"
),
)streamable_http_client() furnizează fluxurile pentru ClientSession; apelul initialize() negociază sesiunea înainte de call_tool(). Orchestratorul verifică explicit isError, existența conținutului și forma textului.
În A2A, cardul este descoperit înaintea apelului. Cererea folosește JSON-RPC 2.0 și message/send. Funcția find_verification() traversează răspunsul deoarece textul agentului poate fi împachetat într-un mesaj sau într-un artifact al task-ului A2A.
Această verificare este o poartă de acceptare, nu doar logging. Este același principiu care trebuie urmărit când construiești evaluări pentru agenți AI cu tool-uri: succesul trebuie legat de o condiție măsurabilă, nu de faptul că agentul a produs text.
Punct de control: dacă editorul semnalează BaseAgent ca model Pydantic, nu muta URL-urile în atribute create dinamic în __init__; câmpurile tipizate mcp_url și agent_card_url sunt declarate în clasă tocmai pentru validarea ADK.
Pasul 5 — Rulăm orchestratorul prin Runner și salvăm atomic rezultatul
De ce: Runner oferă ciclul normal de execuție ADK și o sesiune identificabilă. Fișierul final este scris numai după ce agentul produce evenimentul acceptat.
Fișierul orchestrator/run.py:
from __future__ import annotations
import asyncio
import json
from pathlib import Path
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.genai import types
from orchestrator.agent import root_agent
APP_NAME = "inventory_verification"
USER_ID = "local-operator"
SESSION_ID = "inventory-run-001"
PROJECT_ROOT = Path(__file__).resolve().parent.parent
REQUEST_PATH = PROJECT_ROOT / "data" / "request.json"
RESULT_PATH = PROJECT_ROOT / "runs" / "result.json"
async def main() -> None:
if not REQUEST_PATH.is_file():
raise SystemExit(f"Missing request file: {REQUEST_PATH}")
request = json.loads(REQUEST_PATH.read_text(encoding="utf-8"))
session_service = InMemorySessionService()
await session_service.create_session(
app_name=APP_NAME,
user_id=USER_ID,
session_id=SESSION_ID,
)
runner = Runner(
agent=root_agent,
app_name=APP_NAME,
session_service=session_service,
)
final_text: str | None = None
message = types.Content(
role="user",
parts=[
types.Part(
text=json.dumps(
request,
ensure_ascii=False,
separators=(",", ":"),
sort_keys=True,
)
)
],
)
async for event in runner.run_async(
user_id=USER_ID,
session_id=SESSION_ID,
new_message=message,
):
if event.author != root_agent.name or not event.content:
continue
for part in event.content.parts or []:
if part.text:
final_text = part.text
if final_text is None:
raise SystemExit("The orchestrator produced no final result")
result = json.loads(final_text)
if result.get("accepted") is not True:
raise SystemExit("The final result was not accepted")
RESULT_PATH.parent.mkdir(parents=True, exist_ok=True)
temporary_path = RESULT_PATH.with_suffix(".json.tmp")
temporary_path.write_text(
json.dumps(
result,
ensure_ascii=False,
indent=2,
sort_keys=True,
)
+ "n",
encoding="utf-8",
)
temporary_path.replace(RESULT_PATH)
print(
json.dumps(
{
"accepted": result["accepted"],
"count": result["manifest"]["count"],
"result_file": str(RESULT_PATH),
},
ensure_ascii=False,
sort_keys=True,
)
)
if __name__ == "__main__":
asyncio.run(main())Scrierea folosește un fișier temporar urmat de replace(). Un proces întrerupt înaintea înlocuirii nu lasă un result.json parțial pe care altă automatizare l-ar putea interpreta drept rezultat valid.
Rulează orchestratorul când cele două servicii sunt active:
source .venv/bin/activate
python -m orchestrator.runIeșirea așteptată este:
{"accepted": true, "count": 3, "result_file": "/home/UTILIZATOR/labs/adk-a2a-mcp/runs/result.json"}Calea include numele real al utilizatorului. Digestul complet se află în runs/result.json.
Punct de control: dacă nu apare fișierul, citește mai întâi excepția orchestratorului; nu crea manual result.json, deoarece tocmai absența lui împiedică acceptarea unui schimb incomplet.
Pasul 6 — Automatizăm pornirea, oprirea și inspecția logurilor
De ce: testul trebuie să pornească aceleași servicii în aceeași ordine și să refuze porturile ocupate. Scriptul păstrează PID-urile proceselor lansate, fără să oprească procese străine.
Fișierul scripts/start.sh:
#!/usr/bin/env bash
set -euo pipefail
PROJECT_ROOT="$HOME/labs/adk-a2a-mcp"
cd "$PROJECT_ROOT"
source .venv/bin/activate
mkdir -p logs runs
rm -f runs/mcp.pid runs/a2a.pid
port_is_open() {
python - "$1" <<'PY'
import socket
import sys
port = int(sys.argv[1])
with socket.socket() as sock:
sock.settimeout(0.2)
raise SystemExit(
0 if sock.connect_ex(("127.0.0.1", port)) == 0 else 1
)
PY
}
wait_for_port() {
local port="$1"
local attempts=50
for ((i = 1; i <= attempts; i++)); do
if port_is_open "$port"; then
return 0
fi
sleep 0.2
done
echo "Portul $port nu a devenit disponibil." >&2
return 1
}
if port_is_open 8000 || port_is_open 8001; then
echo "Portul 8000 sau 8001 este deja ocupat." >&2
exit 1
fi
python -m services.mcp_service
>logs/mcp.log 2>&1 &
echo "$!" >runs/mcp.pid
uvicorn remote_agent.server:app
--host 127.0.0.1
--port 8001
>logs/a2a.log 2>&1 &
echo "$!" >runs/a2a.pid
cleanup_on_error() {
"$PROJECT_ROOT/scripts/stop.sh" || true
}
trap cleanup_on_error ERR
wait_for_port 8000
wait_for_port 8001
curl --fail --silent
http://127.0.0.1:8001/.well-known/agent-card.json
>/dev/null
trap - ERR
echo "MCP=127.0.0.1:8000 și A2A=127.0.0.1:8001 sunt active."Fișierul scripts/stop.sh:
#!/usr/bin/env bash
set -euo pipefail
PROJECT_ROOT="$HOME/labs/adk-a2a-mcp"
for name in a2a mcp; do
pid_file="$PROJECT_ROOT/runs/$name.pid"
if [[ ! -f "$pid_file" ]]; then
continue
fi
pid="$(cat "$pid_file")"
if [[ "$pid" =~ ^[0-9]+$ ]] && kill -0 "$pid" 2>/dev/null; then
kill "$pid"
for _ in {1..30}; do
if ! kill -0 "$pid" 2>/dev/null; then
break
fi
sleep 0.1
done
fi
rm -f "$pid_file"
done
echo "Procesele înregistrate de laborator au fost oprite."Activează execuția și pornește serviciile:
chmod 750 scripts/start.sh scripts/stop.sh
./scripts/start.sh
tail -n 20 logs/mcp.log
tail -n 20 logs/a2a.logRezultatul de control este:
MCP=127.0.0.1:8000 și A2A=127.0.0.1:8001 sunt active.set -euo pipefail oprește scriptul la o comandă eșuată, la o variabilă nedefinită sau la o eroare ascunsă într-un pipeline. Verificarea porturilor se face înaintea lansării, iar trap curăță primul proces dacă al doilea nu pornește.
Punct de control: dacă scriptul spune că un port este ocupat, identifică procesul cu ss -ltnp | grep -E ':8000|:8001'; nu îl opri până nu confirmi că îți aparține.
Pasul 7 — Recalculăm dovada fără să avem încredere în agent
De ce: testul final nu trebuie să considere câmpul verified suficient. El reconstruiește forma canonică a manifestului și compară independent ambele digesturi.
Fișierul tests/test_result.py:
from __future__ import annotations
import hashlib
import json
from pathlib import Path
from typing import Any
PROJECT_ROOT = Path(__file__).resolve().parent.parent
RESULT_PATH = PROJECT_ROOT / "runs" / "result.json"
def canonical_json(value: Any) -> str:
return json.dumps(
value,
ensure_ascii=False,
separators=(",", ":"),
sort_keys=True,
)
def test_result_is_verified_end_to_end() -> None:
assert RESULT_PATH.is_file()
result = json.loads(RESULT_PATH.read_text(encoding="utf-8"))
manifest = result["manifest"]
local_digest = hashlib.sha256(
canonical_json(manifest).encode("utf-8")
).hexdigest()
assert result["accepted"] is True
assert manifest["schema"] == "cyberai.inventory.v1"
assert manifest["count"] == 3
assert len(manifest["items"]) == 3
assert result["sha256"] == local_digest
assert result["remote"]["verified"] is True
assert result["remote"]["received_sha256"] == local_digest
assert result["remote"]["recomputed_sha256"] == local_digestFișierul scripts/smoke.sh:
#!/usr/bin/env bash
set -euo pipefail
PROJECT_ROOT="$HOME/labs/adk-a2a-mcp"
cd "$PROJECT_ROOT"
source .venv/bin/activate
cleanup() {
"$PROJECT_ROOT/scripts/stop.sh" || true
}
trap cleanup EXIT
rm -f runs/result.json runs/result.json.tmp
./scripts/start.sh
python -m orchestrator.run
python -m pytest -q tests/test_result.py
jq '{
accepted,
count: .manifest.count,
remote_verified: .remote.verified
}' runs/result.jsonRulează testul complet:
chmod 750 scripts/smoke.sh
./scripts/smoke.shUn exemplu de ieșire reușită este:
MCP=127.0.0.1:8000 și A2A=127.0.0.1:8001 sunt active.
{"accepted": true, "count": 3, "result_file": "/home/UTILIZATOR/labs/adk-a2a-mcp/runs/result.json"}
1 passed
{
"accepted": true,
"count": 3,
"remote_verified": true
}
Procesele înregistrate de laborator au fost oprite.Orele, durata Pytest și calea utilizatorului pot diferi. Condițiile stabile sunt un test trecut, count: 3 și cele două valori booleene true.
Poți demonstra că poarta funcționează modificând temporar o copie a rezultatului, dar nu este necesar să falsifici traficul sau să interceptezi conexiunea. Testul independent este suficient pentru acceptarea normală; scenariile de alterare controlată aparțin unei suite de teste separate.
Punct de control: după rulare, ss -ltn | grep -E ':8000|:8001' nu trebuie să mai găsească listener-ele laboratorului, deoarece trap execută stop.sh.
Cinci defecțiuni specifice acestui traseu
McpError: Streamable HTTP error: Connection refused
Cauza este absența serviciului MCP pe 127.0.0.1:8000 sau terminarea sa imediată după lansare. Citește logul și repornește întregul traseu:
cat logs/mcp.log
./scripts/stop.sh
./scripts/start.shNu schimba URL-ul în SSE; proiectul este scris pentru transportul Streamable HTTP.
MCP build_manifest returned an error
Unealta a respins cererea. Cauzele implementate explicit sunt lista goală, un name ori version gol sau două componente cu același nume. Validează fișierul și inspectează valorile:
jq -e '
.items | length > 0 and
all(.[]; (.name | length > 0) and (.version | length > 0))
' data/request.json
jq -r '.items[].name' data/request.json | sort | uniq -dPrima comandă trebuie să întoarcă true, iar a doua nu trebuie să afișeze nimic.
Agent Card does not contain url
Orchestratorul a primit un document care nu corespunde cardului așteptat de versiunea fixată sau a fost direcționat către alt serviciu. Verifică endpoint-ul:
curl --fail --silent
http://127.0.0.1:8001/.well-known/agent-card.json |
jq -e '.url'Nu introduce un endpoint RPC fix în cod ca să ascunzi problema; descoperirea prin Agent Card este parte a testului.
A2A response contains no verification artifact
Apelul A2A a răspuns, dar arborele result nu conține JSON-ul produs de agent. Verifică logs/a2a.log și confirmă că serverul încarcă agentul din acest proiect:
grep -E 'remote_digest_verifier|ERROR|Traceback' logs/a2a.log
cat runs/a2a.pid
ps -fp "$(cat runs/a2a.pid)"Dacă ai schimbat codul, oprește și repornește Uvicorn; scriptul nu activează reload automat.
Remote verification does not match the MCP result
Răspunsul există, dar digestul primit, digestul recalculat, numărul de elemente sau verified nu coincide cu rezultatul MCP. Păstrează logurile și rezultatul pentru analiză, apoi rulează testul independent:
jq '{
local: .sha256,
received: .remote.received_sha256,
recomputed: .remote.recomputed_sha256,
verified: .remote.verified
}' runs/result.json
python -m pytest -q tests/test_result.pyOrchestratorul nu trebuie modificat să accepte doar unul dintre digesturi. Nepotrivirea este exact situația pe care proiectul trebuie să o blocheze.
Curățăm laboratorul fără să atingem alte procese
Oprește procesele înregistrate:
cd "$HOME/labs/adk-a2a-mcp"
./scripts/stop.sh
rm -f runs/result.json runs/result.json.tmp
rm -f logs/mcp.log logs/a2a.logPentru eliminarea completă a laboratorului, ieși din director înainte de ștergere:
deactivate 2>/dev/null || true
cd "$HOME/labs"
rm -rf -- "$HOME/labs/adk-a2a-mcp"Avertisment: ultima comandă șterge definitiv mediul virtual, sursele, logurile și rezultatele proiectului. Verifică exact calea înainte să o rulezi. Nu folosi o variabilă goală și nu executa ștergerea din alt proiect.
Nu există pachete de sistem dedicate de dezinstalat în afară de utilitarele generale curl și jq; tutorialul nu recomandă eliminarea lor, deoarece pot fi folosite și de alte aplicații.
Lista de acceptanță pentru schimbul dintre servicii
- ☐
pip checkraportează că dependențele instalate nu sunt incompatibile. - ☐ Serviciul MCP ascultă numai pe
127.0.0.1:8000. - ☐ Agentul A2A ascultă numai pe
127.0.0.1:8001. - ☐ Agent Card-ul conține numele verificatorului și URL-ul său.
- ☐
build_manifestsortează componentele și respinge duplicatele. - ☐ Agentul remote recalculează SHA-256 din manifest, nu copiază verdictul local.
- ☐ Orchestratorul compară digestul MCP cu ambele digesturi A2A.
- ☐
runs/result.jsoneste scris numai dupăaccepted: true. - ☐ Pytest afișează
1 passed. - ☐ Scriptul de test oprește ambele servicii la ieșire.
Următorul prag realist este înlocuirea loopback-ului cu două gazde de laborator și protejarea A2A și MCP prin TLS și autentificare între servicii. A doua extensie este propagarea unui identificator de corelare în logurile MCP, A2A și ADK, astfel încât aceeași execuție să poată fi urmărită fără a salva conținut sensibil. A treia este o suită de evaluare care rulează cereri valide, duplicate și manifesturi alterate și măsoară separat acceptările corecte și respingerile corecte.
Întrebări frecvente
Este necesară o cheie Gemini pentru acest proiect?
Nu. Atât orchestratorul, cât și verificatorul sunt agenți ADK determinați prin BaseAgent și nu apelează un model generativ.
De ce sunt necesare atât MCP, cât și A2A?
MCP expune funcția build_manifest ca unealtă, iar A2A transportă rezultatul către un agent aflat în alt serviciu. Ele acoperă granițe diferite ale proiectului.
SHA-256 autentifică serviciul remote?
Nu. Digestul detectează modificarea reprezentării canonice, dar nu dovedește identitatea expeditorului; o implementare de rețea are nevoie și de TLS, autentificare și autorizare.
Pot expune porturile 8000 și 8001 în rețeaua biroului?
Nu în configurația prezentată. Laboratorul leagă ambele servicii de 127.0.0.1, iar expunerea într-o rețea cere controale suplimentare de transport și identitate.
