microsoft · MehakBindra · Apr 25, 2026 · Apr 23, 2026 · Apr 23, 2026 · Apr 23, 2026
diff --git a/examples/a2a-test/README.md b/examples/a2a-test/README.md
@@ -1,14 +1,117 @@
-# Sample: A2A Client and Server
-a sample showcasing an a2a server / client
+# Sample: Two Teams bots relaying questions via A2A + Adaptive Cards
 
+Two symmetric Teams bots, Alice and Bob, each backed by an LLM agent. The user DMs one of them; the LLM decides whether to answer directly or forward the question to the other bot over the A2A protocol.
 
-Open up the client, and send a message:
+This sample demonstrates:
+
+- **LLM-driven peer routing** — each bot's agent reads the other's A2A `AgentCard.description` (fetched lazily via `A2ACardResolver`) and uses that to decide whether to forward.
+- **Human-in-the-loop via Adaptive Cards** — when a peer asks, the answering bot pushes an ask-card to its human operator's 1:1; the operator types a reply and submits.
+- **Async reply, folded back into chat** — the answer comes back over A2A and is delivered both as a reply card and as a `[peer update]` note injected into the user's LLM session, so the next turn's model sees it as context.
+
+## Flow
+
+```mermaid
+sequenceDiagram
+    actor UA as User-A
+    participant A as Alice (LLM agent)
+    participant B as Bob (LLM agent)
+    actor OB as Operator-B
+
+    UA->>A: "how do I scale my postgres database?"
+    Note over A: LLM reads Bob's AgentCard ("backend & infra"), picks tool:<br/>send_to_peer("bob", "how do I scale my postgres database?")<br/>stash awaiting_reply[qid] = User-A conv
+    A->>B: A2A ask {qid, question, sender, reply_url=Alice}
+    A-->>UA: streamed reply ("Asked Bob, will let you know…")
+    Note over B: validate reply_url ∈ allowlist<br/>stash inbound_asks[qid] = {reply_url, sender, question}
+    B->>OB: push ask card
+    OB->>B: submit "use read replicas + pgbouncer" (carries qid)
+    Note over B: pop inbound_asks[qid] → trusted reply_url
+    B->>A: A2A reply {qid, answer, responder}
+    Note over A: pop awaiting_reply[qid]<br/>inject "[peer update] Bob replied: 'use read replicas + pgbouncer'…" into User-A's session
+    A->>UA: push reply card
+```
+
+## Files
+
+**Entry points** — start here.
+- `src/bot_a.py` — Alice. Teams `/api/messages` and A2A `/a2a` share port **3978**. Edit the `DESCRIPTION` constant to set Alice's expertise; this becomes her A2A AgentCard description that Bob's LLM reads to decide when to forward.
+- `src/bot_b.py` — Bob. Same layout on port **3979**. Same `DESCRIPTION` knob for Bob.
+
+**LLM agent**
+- `src/agent.py` — `BotAgent` builds the `agent_framework` `Agent`, lazily fetches peer A2A cards via `A2ACardResolver`, and exposes `get_agent()`, `session_for(conv_id)`, and `record_peer_reply(...)` for the bot file to use.
+
+**A2A layer**
+- `src/a2a_executor.py` — A2A server dispatch: `ask` → validate `reply_url`, stash, push card to operator; `reply` → push card to the original user and call `on_peer_reply`.
+- `src/a2a_server.py` — `make_a2a_app(..., allowed_peer_urls=..., on_peer_reply=...)` wraps the executor in `A2AStarletteApplication`.
+- `src/a2a_client.py` — `send_a2a(peer_url, data)` one-shot sender, plus `is_allowed_peer(url, allowed)` for origin-based peer URL validation.
+- `src/messages.py` — `AskMessage` / `ReplyMessage` Pydantic models with a `kind` discriminator.
+
+**Cards & state**
+- `src/cards.py` — `ask_card(sender, question, qid)` (submit carries only qid), `reply_card(...)`.
+- `src/state.py` — `BotState` (operator conversation, outbound asks awaiting a reply, inbound asks awaiting an operator).
+
+## Operator model
+
+Each bot remembers the last **1:1** Teams conversation that messaged it (`state.operator_conv_id`). Incoming asks are pushed into that conversation.
+
+## Peer authorization
+
+The `reply_url` check in `is_allowed_peer` is a **demo-only** stand-in for authorization: a peer is trusted because its URL matches a configured origin. Production A2A should verify the caller's identity with a bearer token signed by an IdP or mTLS, not a self-declared URL.
+
+## Configuration
+
+Create `.env` in `examples/a2a-test/`:
+
+```dotenv
+# Shared — your Microsoft tenant
+TENANT_ID=<your-tenant-id>
+
+# Azure OpenAI — used by both bots' LLM
+AZURE_OPENAI_API_KEY=<key>
+AZURE_OPENAI_ENDPOINT=<endpoint>
+AZURE_OPENAI_MODEL=<deployment-name>
+
+# Bot A (Alice) — Teams app registration
+BOT_A_CLIENT_ID=<alice-client-id>
+BOT_A_CLIENT_SECRET=<alice-client-secret>
+
+# Bot B (Bob) — Teams app registration
+BOT_B_CLIENT_ID=<bob-client-id>
+BOT_B_CLIENT_SECRET=<bob-client-secret>
+
+# Optional — ports and A2A peer URLs (defaults shown)
+# BOT_A_HOST=localhost
+# BOT_A_PORT=3978
+# BOB_A2A_URL=http://localhost:3979/a2a/
+# BOT_B_HOST=localhost
+# BOT_B_PORT=3979
+# ALICE_A2A_URL=http://localhost:3978/a2a/
+```
+
+Each bot needs its **own** Teams app registration so DMs route to the right bot.
+
+## Run
+
+Two terminals from `examples/a2a-test/`:
+
+```bash
+uv run python src/bot_a.py   # Alice — Teams + A2A on 3978
+uv run python src/bot_b.py   # Bob   — Teams + A2A on 3979
+```
+
+> ⚠ **DM each bot once before relaying.** The operator's conversation id is captured from the first Teams message the bot receives. If a peer ask arrives before its target has been DM'd, the target will log `no operator conversation; ask not pushed` and the card won't appear anywhere.
+
+### Try it
+
+With both bots DM'd at least once, try this transcript against Alice:
 
 ```
-C: What's the weather like?
-S: Could you please specify the location for which you'd like to know the weather?
-C: London
-S: The weather in London is sunny
-C: What's the weather like in Tokyo?
-S: The weather in Tokyo is sunny
-```
+You → Alice: how do I scale my postgres database?
+Alice → You: Asked Bob, will let you know…
+   (Bob's operator gets an ask card, types "use read replicas + pgbouncer", submits)
+Alice → You: [reply card] Bob says: use read replicas + pgbouncer
+...... conversation ...
+You → Alice: how do I scale my postgres database again, i forgot  ?
+Alice → You:  ... Bob’s short recommendation earlier was: read replicas + PgBouncer. ...
+```
+
+The bots are symmetric — DM Bob with a UX question(eg. What's the best way to design a website?) and the same flow runs the other way (Bob's LLM forwards to Alice).
diff --git a/examples/a2a-test/pyproject.toml b/examples/a2a-test/pyproject.toml
@@ -1,21 +1,22 @@
 [project]
-name = "a2a"
+name = "a2a-test"
 version = "0.1.0"
-description = "sample showcasing a2a server and client"
+description = "Two Teams bots talking to each other via the official a2a-sdk, exchanging Adaptive Cards"
 readme = "README.md"
 requires-python = ">=3.12,<3.15"
 dependencies = [
     "dotenv>=0.9.9",
     "microsoft-teams-apps",
-    "microsoft-teams-a2a",
-    "microsoft-teams-ai",
-    "microsoft-teams-openai",
     "microsoft-teams-common",
+    "microsoft-teams-cards",
+    "a2a-sdk[core,http-server]>=0.3.7",
+    "agent-framework-core",
+    "agent-framework-openai",
+    "uvicorn>=0.30",
+    "httpx>=0.27",
 ]
 
 [tool.uv.sources]
 microsoft-teams-apps = { workspace = true }
-microsoft-teams-ai = { workspace = true }
-microsoft-teams-a2a = { workspace = true }
-microsoft-teams-openai = { workspace = true }
 microsoft-teams-common = { workspace = true }
+microsoft-teams-cards = { workspace = true }
diff --git a/examples/a2a-test/src/a2a_client.py b/examples/a2a-test/src/a2a_client.py
@@ -0,0 +1,64 @@
+"""
+Copyright (c) Microsoft Corporation. All rights reserved.
+Licensed under the MIT License.
+"""
+
+import uuid
+from typing import Any
+from urllib.parse import urlsplit
+
+import httpx
+from a2a.client import A2ACardResolver, ClientConfig, ClientFactory
+from a2a.types import DataPart, Message, Part, Role
+
+# Outbound A2A helpers + peer URL allowlist check.
+
+_DEFAULT_PORTS = {"http": 80, "https": 443}
+
+
+def _origin(url: str) -> tuple[str, str, int] | None:
+    try:
+        parts = urlsplit(url)
+    except ValueError:
+        return None
+    scheme = parts.scheme.lower()
+    host = (parts.hostname or "").lower()
+    if not scheme or not host:
+        return None
+    port = parts.port if parts.port is not None else _DEFAULT_PORTS.get(scheme, 0)
+    return (scheme, host, port)
+
+
+def is_allowed_peer(url: str, allowed: list[str]) -> bool:
+    # Demo-only stand-in for authorization: we trust a peer because its
+    # reply_url matches a configured origin. Production A2A should verify the
+    # caller's identity (e.g. bearer token signed by an IdP, or mTLS) rather
+    # than trusting a self-declared URL. Match by scheme/host/port so a
+    # trailing slash or default port doesn't flip a valid peer to invalid.
+    target = _origin(url)
+    if target is None:
+        return False
+    for candidate in allowed:
+        candidate_origin = _origin(candidate)
+        if candidate_origin is not None and candidate_origin == target:
+            return True
+    return False
+
+
+async def send_a2a(peer_url: str, data: dict[str, Any]) -> None:
+    # Resolve the peer's agent card, build an a2a-sdk client, and fire a
+    # single DataPart-carrying message. We drain the response stream
+    # without reading it — the peer only sends an `ack`; any "real"
+    # answer comes later as a separate inbound A2A call back to us.
+    async with httpx.AsyncClient(timeout=60.0) as http_client:
+        peer_card = await A2ACardResolver(httpx_client=http_client, base_url=peer_url).get_agent_card()
+        factory = ClientFactory(ClientConfig(httpx_client=http_client, streaming=True))
+        client = factory.create(peer_card)
+
+        request = Message(
+            message_id=str(uuid.uuid4()),
+            role=Role.user,
+            parts=[Part(root=DataPart(data=data))],
+        )
+        async for _ in client.send_message(request):
+            pass
diff --git a/examples/a2a-test/src/a2a_executor.py b/examples/a2a-test/src/a2a_executor.py
@@ -0,0 +1,138 @@
+"""
+Copyright (c) Microsoft Corporation. All rights reserved.
+Licensed under the MIT License.
+"""
+
+import logging
+import uuid
+from typing import Callable, Optional
+
+from a2a.server.agent_execution.agent_executor import AgentExecutor
+from a2a.server.agent_execution.context import RequestContext
+from a2a.server.events.event_queue import EventQueue
+from a2a.types import (
+    DataPart,
+    Message,
+    Part,
+    Role,
+    TaskState,
+    TaskStatus,
+    TaskStatusUpdateEvent,
+)
+from a2a_client import is_allowed_peer
+from cards import ask_card, reply_card
+from messages import A2AMessage, A2AMessageAdapter, AskMessage, ReplyMessage
+from microsoft_teams.apps import App
+from pydantic import ValidationError
+from state import BotState
+
+OnPeerReply = Callable[[str, str, str, str], None]
+# (user_conv_id, responder, question, answer) -> None
+
+# A2A server-side dispatch. Reads the incoming `DataPart`, branches on
+# `data.kind` (`ask` vs `reply`), updates `BotState`, and builds the Teams
+# card locally from the payload.
+
+logger = logging.getLogger(__name__)
+
+
+def parse_a2a_message(message: Optional[Message]) -> Optional[A2AMessage]:
+    # A2A messages can carry multiple parts; this sample only uses one
+    # DataPart per message. Validate it against the discriminated union so
+    # the executor never handles a raw dict.
+    if message is None:
+        return None
+    for part in message.parts:
+        if isinstance(part.root, DataPart):
+            try:
+                return A2AMessageAdapter.validate_python(part.root.data)
+            except ValidationError as e:
+                logger.warning("invalid a2a message: %s", e)
+                return None
+    return None
+
+
+class AskReplyExecutor(AgentExecutor):
+    def __init__(
+        self,
+        teams_app: App,
+        state: BotState,
+        allowed_peer_urls: list[str],
+        on_peer_reply: Optional[OnPeerReply] = None,
+    ) -> None:
+        self._teams_app = teams_app
+        self._state = state
+        self._allowed_peer_urls = allowed_peer_urls
+        self._on_peer_reply = on_peer_reply
+
+    async def execute(self, context: RequestContext, event_queue: EventQueue) -> None:
+        task_id = context.task_id or str(uuid.uuid4())
+        context_id = context.context_id or str(uuid.uuid4())
+        message = parse_a2a_message(context.message)
+
+        if isinstance(message, AskMessage):
+            await self._on_ask(message)
+        elif isinstance(message, ReplyMessage):
+            await self._on_reply(message)
+
+        # A2A tasks need a terminal status event to close out. Our "real"
+        # response (if any) flows later as a separate inbound A2A message
+        # from the peer, so we just ack this one and finish.
+        ack = Message(
+            message_id=str(uuid.uuid4()),
+            role=Role.agent,
+            parts=[Part(root=DataPart(data={"kind": "ack"}))],
+        )
+        await event_queue.enqueue_event(
+            TaskStatusUpdateEvent(
+                task_id=task_id,
+                context_id=context_id,
+                status=TaskStatus(state=TaskState.completed, message=ack),
+                final=True,
+            )
+        )
+
+    async def cancel(self, context: RequestContext, event_queue: EventQueue) -> None:
+        await event_queue.enqueue_event(
+            TaskStatusUpdateEvent(
+                task_id=context.task_id or str(uuid.uuid4()),
+                context_id=context.context_id or str(uuid.uuid4()),
+                status=TaskStatus(state=TaskState.canceled),
+                final=True,
+            )
+        )
+
+    async def _on_ask(self, msg: AskMessage) -> None:
+        # Peer is asking us a question. Stash routing by qid and push the
+        # ask card to our operator.
+        logger.info("[%s] received ask qid=%s from %s", self._state.name, msg.qid, msg.sender)
+        conv_id = self._state.operator_conv_id
+        if not conv_id:
+            logger.warning("[%s] no operator conversation; ask not pushed", self._state.name)
+            return
+        # Validate reply_url before we stash it or push a card tied to it.
+        if not is_allowed_peer(msg.reply_url, self._allowed_peer_urls):
+            logger.warning(
+                "[%s] rejecting ask qid=%s: reply_url %r not in allowlist", self._state.name, msg.qid, msg.reply_url
+            )
+            return
+        self._state.inbound_asks[msg.qid] = {
+            "reply_url": msg.reply_url,
+            "sender": msg.sender,
+            "question": msg.question,
+        }
+        await self._teams_app.send(conv_id, ask_card(sender=msg.sender, question=msg.question, qid=msg.qid))
+
+    async def _on_reply(self, msg: ReplyMessage) -> None:
+        # Peer is answering a question we originally sent. Push the reply
+        # card to the user who asked, and let the caller (the LLM agent)
+        # know so it can fold the reply into the user's chat session.
+        pending = self._state.awaiting_reply.pop(msg.qid, None)
+        logger.info("[%s] received reply qid=%s", self._state.name, msg.qid)
+        if not pending:
+            logger.warning("[%s] no awaiting conversation for qid=%s", self._state.name, msg.qid)
+            return
+        card = reply_card(responder=msg.responder, question=pending["question"], answer=msg.answer, qid=msg.qid)
+        await self._teams_app.send(pending["conv_id"], card)
+        if self._on_peer_reply is not None:
+            self._on_peer_reply(pending["conv_id"], msg.responder, pending["question"], msg.answer)