Merge pull request #246 from microsoft/james-dev-agentic-ai

add A2A implementation for Semantic Kernel Multi-Agent

Author: nicoleserafino

README.md

@@ -33,6 +33,7 @@ Welcome to the official repository for the Microsoft AI Agentic Workshop! This r
 
 - **Configurable LLM Backend:** Use the latest Azure OpenAI GPT models (e.g., GPT-4.1, GPT-4o).  
 - **MCP Server Integration:** Advanced tools to enhance agent orchestration and capabilities.  
+- **A2A (Agent-to-Agent) Protocol Support:** Enables strict cross-domain, black-box multi-agent collaboration using [Google's A2A protocol](https://github.com/google-a2a/A2A).
 - **Flexible Agent Architecture:**  
   - Supports single-agent, multi-agent, or reflection-based agents (selectable via `.env`).  
   - Agents can self-loop, collaborate, reflect, or take on dynamic roles as defined in modules.  
@@ -53,6 +54,15 @@ Welcome to the official repository for the Microsoft AI Agentic Workshop! This r
 
 ---  
 
+## 🆕 A2A (Agent-to-Agent) Cross-Domain Demo  
+  
+This repository now supports a strict cross-domain multi-agent scenario using the A2A protocol, enabling message-driven, black-box collaboration between a customer-service agent and a logistics agent.  
+  
+**A2A Example Included:**    
+ See [`agentic_ai/agents/semantic_kernel/multi_agent/a2a`](agentic_ai/agents/semantic_kernel/multi_agent/a2a).  
+  
+---  
+
 ## Contributing  
 
 Please review our [Code of Conduct](./CODE_OF_CONDUCT.md) and [Security Guidelines](./SECURITY.md) before contributing.  

agentic_ai/agents/base_agent.py

@@ -17,7 +17,7 @@ def __init__(self, state_store: Dict[str, Any], session_id: str) -> None:
         self.azure_openai_key = os.getenv("AZURE_OPENAI_API_KEY")  
         self.azure_openai_endpoint = os.getenv("AZURE_OPENAI_ENDPOINT")  
         self.api_version = os.getenv("AZURE_OPENAI_API_VERSION")  
-        self.mcp_server_uri = os.getenv("MCP_SERVER_URI")  
+        self.mcp_server_uri = os.getenv("MCP_SERVER_URI") 
         self.openai_model_name = os.getenv("OPENAI_MODEL_NAME")  
 
         self.session_id = session_id  

agentic_ai/agents/semantic_kernel/multi_agent/a2a/README.md

@@ -0,0 +1,94 @@
+# Cross-Domain Return-Pick-up Scheduling (A2A)
+
+This A2A implementation demonstrates inter-domain communication between agents in different domains. Unlike inter-agent communication within a single domain or application—where participating agents typically have full transparency into each other’s details—cross-domain agent communication enforces strict modularity and abstraction. In cross-domain scenarios, the logic and implementation of each agent system are hidden from one another, and only high-level structured information is exchanged. This approach aligns with Google’s Agent-to-Agent (A2A) protocol principles.  
+  
+### Scenario: Cross-Domain Return Pickup Scheduling  
+  
+In this implementation, an agent within the Contoso Customer Service AI team collaborates with a Logistics Agent to arrange a product return pickup. After verifying the return eligibility, the Customer Service Agent initiates a multi-turn negotiation with the Logistics Agent to schedule a pickup at the customer's address. The process includes:  
+  
+- The Customer Service Agent requesting available pickup slots from the Logistics Agent.  
+- The Logistics Agent responding with a list of available date/time options.  
+- The Customer Service Agent presenting these options to the customer and collecting a preferred slot.  
+- The Customer Service Agent confirming the selected slot with the Logistics Agent, who in turn confirms logistics with the carrier and finalizes the arrangement.  
+- Each communication is handled using high-level, schema-driven A2A messages, with neither agent exposing its internal logic, system details, or direct access to underlying services.  
+  
+---  
+  
+#### Mermaid Flow Diagram  
+
+```mermaid  
+sequenceDiagram  
+    actor Customer  
+    participant CSAgent as Customer Service Agent  
+    participant LogAgent as Logistics Agent  
+  
+    Customer->>CSAgent: Request return for Order #85  
+    CSAgent->>Customer: Verifies eligibility, explains process  
+    CSAgent->>LogAgent: PickupAvailabilityRequest (address, preferences)  
+    LogAgent-->>CSAgent: PickupAvailabilityResponse (list of slots)  
+    CSAgent->>Customer: Presents pickup options  
+    Customer->>CSAgent: Chooses preferred slot  
+    CSAgent->>LogAgent: PickupRequestConfirmation (selected slot)  
+    LogAgent-->>CSAgent: PickupScheduledConfirmation (confirmation details)  
+    CSAgent->>Customer: Confirms pickup details, provides instructions      
+```
+## Running the A2A Demo End-to-End  
+  
+The repo ships three Python modules:  
+  
+| File                      | Purpose                                                   |  
+|---------------------------|-----------------------------------------------------------|  
+| `logistic_mcp.py`         | Internal Logistics **MCP** service (tools & DB)           |  
+| `logistic_a2a_server.py`  | Thin **A2A façade** that wraps the MCP service            |  
+| `multi_agent_a2a.py`      | Contoso **multi-agent** customer-service application      |  
+  
+---  
+  
+### 1. Install Dependencies  
+  
+```bash  
+pip install -r requirements.txt  
+# or manually:  
+pip install a2a-sdk semantic-kernel uvicorn httpx python-dotenv  
+```
+### 2. Prepare your .env	
+ 	
+Create or edit .env in the `agentic_ai\applications` folder:	
+ 	
+```env	
+# ─── Contoso customer-service app ───────────────────────────────	
+AGENT_MODULE="agents.semantic_kernel.multi_agent.a2a.multi_agent_a2a"	
+ 	
+# ─── End-points used by the agents ──────────────────────────────	
+LOGISTIC_MCP_SERVER_URI="http://localhost:8100/sse" # internal Fast-MCP	
+LOGISTICS_A2A_URL="http://localhost:9100" # A2A wrapper	
+```	
+ 	
+Add your usual AZURE_OPENAI_* settings if you have not done so already.	
+ 	
+---	
+ 	
+### 3. Start the Back-End Services (Two Terminals)	
+ 	
+```bash	
+# Terminal ① – internal Logistics MCP	
+python logistic_mcp.py # listens on :8100/sse	
+ 	
+# Terminal ② – A2A façade	
+python logistic_a2a_server.py # listens on :9100 (serves /.well-known/agent.json)	
+```	
+ 	
+---	
+ 	
+### 4. Launch the Contoso Multi-Agent App under `agentic_ai\applications`
+ 	
+```bash	
+./run_application.sh	
+
+```	
+
+The CS agent will now:  
+  
+1. Verify product-return eligibility via the Contoso MCP tools.  
+2. Talk to the Logistics agent through the **single free-text tool** exposed by the A2A server (no JSON payloads needed).  
+3. Keep `taskId` and `contextId` in its session state so subsequent calls continue the same conversation on the Logistics side.  

agentic_ai/agents/semantic_kernel/multi_agent/a2a/logistic_a2a_server.py

@@ -0,0 +1,186 @@
+"""  
+Contoso – Logistics A2A façade  
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~  
+Bridges the internal Fast-MCP logistics tools into the Google A2A  
+protocol.  All business logic continues to live in the MCP service; this  
+wrapper merely acts as a protocol translator.  
+  
+• Listens on  http://0.0.0.0:9100/  
+• Exposes one skill:  return-pick-up scheduling  
+• Streams a single final message per request  
+"""  
+from __future__ import annotations  
+  
+import asyncio  
+import json  
+import logging  
+import os  
+from typing import Any  
+  
+import uvicorn  
+
+from a2a.server.agent_execution import AgentExecutor,RequestContext
+from a2a.server.apps import A2AStarletteApplication  
+from a2a.server.request_handlers import DefaultRequestHandler  
+from a2a.server.tasks import InMemoryTaskStore  
+from a2a.server.events import EventQueue
+from a2a.types import (  
+    AgentCapabilities,  
+    AgentCard,  
+    AgentSkill,  
+    Message,  
+)  
+from a2a.utils import new_agent_text_message, new_task
+from semantic_kernel.agents import ChatCompletionAgent  
+from semantic_kernel.connectors.ai.open_ai import AzureChatCompletion  
+from semantic_kernel.connectors.mcp import MCPSsePlugin  
+from typing import Any, Dict, List, Optional  
+
+from dotenv import load_dotenv  
+# ────────────────────────  Load environment variables  ───────────
+load_dotenv()
+
+# ─────────────────────────  Logging  ──────────────────────────  
+logging.basicConfig(level=logging.INFO)  
+log = logging.getLogger("logistics-a2a")  
+
+# ────────────────────────  Agent State Store  ────────────────────────────
+AGENT_STATE_STORE: Dict[str, Any] = {}
+  
+# ─────────────────────────  Environment  ──────────────────────  
+MCP_URI = os.getenv("LOGISTIC_MCP_SERVER_URI", "http://localhost:8100/sse")  
+AZ_DEPLOYMENT = os.getenv("AZURE_OPENAI_CHAT_DEPLOYMENT")  
+  
+# ───────────────────────  Build SK Logistics agent  ───────────  
+async def build_sk_logistics_agent() -> ChatCompletionAgent:  
+    """  
+    Creates the Semantic-Kernel ChatCompletionAgent and opens the SSE  
+    connection to the Fast-MCP server.  
+    """  
+    logistic_plugin = MCPSsePlugin(  
+        name="LogisticMCP",  
+        description="Logistics MCP plugin",  
+        url=MCP_URI,  
+        headers={"Content-Type": "application/json"},  
+        timeout=30,  
+    )  
+    await logistic_plugin.connect()  
+  
+    instructions = (  
+        "You are the Logistics AI agent responsible for arranging product-return "  
+        "pick-ups."  
+        "Supported request types:\n"  
+        "  • availability_request\n"  
+        "  • schedule_pickup\n"  
+        "  • cancel_request\n"  
+    )  
+  
+    agent = ChatCompletionAgent(  
+        name="logistics_sk_agent",  
+        service=AzureChatCompletion(deployment_name=AZ_DEPLOYMENT),  
+        instructions=instructions,  
+        plugins=[logistic_plugin],  
+    )  
+    return agent  
+  
+  
+# ────────────────────────  Agent Executor  ─────────────────────  
+class LogisticsA2AExecutor(AgentExecutor):  
+    """  
+    Thin wrapper that forwards the raw JSON payload to a Semantic-Kernel  
+    agent which, in turn, calls the Logistics MCP tools.  
+  
+    The SK agent is created lazily on first use so we do not need an  
+    event-loop during __init__.  
+    """  
+  
+    def __init__(self) -> None:  
+        self._agent: ChatCompletionAgent | None = None  
+        self._agent_lock = asyncio.Lock()  # guards one-time initialisation  
+  
+    async def _get_agent(self) -> ChatCompletionAgent:  
+        if self._agent is None:  
+            async with self._agent_lock:  
+                if self._agent is None:  # double-checked  
+                    self._agent = await build_sk_logistics_agent()  
+        return self._agent  
+  
+    async def execute(      # type: ignore[override]  
+        self,  
+        context: RequestContext,  
+        event_queue: EventQueue,  
+    ) -> None:  
+        try:  
+            agent = await self._get_agent()  
+            query = context.get_user_input()
+            task = context.current_task
+            if not task:
+                task = new_task(context.message)
+                await event_queue.enqueue_event(task)
+            print("received contextid", task.contextId)
+            #get thread from session store
+            thread = AGENT_STATE_STORE.get(task.contextId, {})
+            # Retrieve user's raw JSON payload (1st text part)  
+  
+            # Forward request to the SK logistics agent  
+            if thread:
+                response = await agent.get_response(messages=query, thread=thread) 
+            else:
+                response = await agent.get_response(messages=query)
+            response_content = str(response.content) 
+            # Update the thread in the session store
+            AGENT_STATE_STORE[task.contextId] = response.thread if response.thread else {}
+  
+            # Ensure the answer is valid JSON  
+  
+            await event_queue.enqueue_event(  
+                new_agent_text_message(response_content,  task.contextId,
+                            task.id) 
+            )  
+  
+        except Exception as exc:  # pragma: no cover  
+            logging.exception("LogisticsA2AExecutor error")  
+            event_queue.enqueue_event(  
+                new_agent_text_message(f"ERROR: {exc}")  
+            )  
+  
+    async def cancel(       # type: ignore[override]  
+        self,  
+        context: RequestContext,  
+        event_queue: EventQueue,  
+    ) -> None:  
+        event_queue.enqueue_event(  
+            new_agent_text_message("Cancellation not supported", is_final=True)  
+        )  
+  
+  
+# ──────────────────────────  Agent Card  ───────────────────────  
+skill = AgentSkill(  
+    id="return_pickup",  
+    name="Return pick-up scheduling",  
+    description="Provides slots, books, looks up or cancels product-return pick-ups.",  
+    tags=["logistics", "return"],  
+)  
+  
+PUBLIC_CARD = AgentCard(  
+    name="Contoso Logistics Agent",  
+    description="Cross-domain logistics service for product returns.",  
+    url="http://0.0.0.0:9100/",  
+    version="1.0.0",  
+    defaultInputModes=["text"],  
+    defaultOutputModes=["text"],  
+    capabilities=AgentCapabilities(streaming=True),  
+    skills=[skill],  
+)  
+  
+# ─────────────────────────  Run server  ────────────────────────  
+def main() -> None:  
+    handler = DefaultRequestHandler(  
+        agent_executor=LogisticsA2AExecutor(), task_store=InMemoryTaskStore()  
+    )  
+    app = A2AStarletteApplication(agent_card=PUBLIC_CARD, http_handler=handler)  
+    uvicorn.run(app.build(), host="0.0.0.0", port=9100)  
+  
+  
+if __name__ == "__main__":  
+    main()