wehub/microsoft--ai-agents-for-beginners
1
0
Fork 0
Code Issues Pull Requests Actions Deployments Models Agent Notes Packages Projects Releases Wiki Activity

chore: import upstream snapshot with attribution

Browse Source
This commit is contained in:
wehub-resource-sync
2026-07-13 12:59:43 +08:00
commit c79da9cdf9
9350 changed files with 1725465 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
11-agentic-protocols/code_samples/11-a2a-agent-framework.ipynb
+238
View File
@@ -0,0 +1,238 @@
{
"cells": [
{
"cell_type": "markdown",
"metadata": {},
"source": [
"# Lesson 11 - Agent-to-Agent (A2A) Protocol"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"## Setup"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"%pip install agent-framework azure-ai-projects azure-identity python-dotenv"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"import os\n",
"import dotenv\n",
"from agent_framework import tool, AgentResponseUpdate, WorkflowBuilder\n",
"from agent_framework.foundry import FoundryChatClient\n",
"from azure.identity import DefaultAzureCredential\n",
"\n",
"dotenv.load_dotenv()\n",
"\n",
"endpoint = os.getenv(\"AZURE_AI_PROJECT_ENDPOINT\")\n",
"deployment_name = os.getenv(\"AZURE_AI_MODEL_DEPLOYMENT_NAME\")\n",
"\n",
"missing = [k for k, v in {\n",
" \"AZURE_AI_PROJECT_ENDPOINT\": endpoint,\n",
" \"AZURE_AI_MODEL_DEPLOYMENT_NAME\": deployment_name\n",
"}.items() if not v]\n",
"\n",
"if missing:\n",
" raise ValueError(\n",
" f\"Missing required environment variables: {', '.join(missing)}. \"\n",
" \"Please set them as environment variables (e.g., in your .env file or shell environment).\"\n",
" )"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"# Create the Microsoft Foundry client\n",
"client = FoundryChatClient(\n",
" project_endpoint=endpoint,\n",
" model=deployment_name,\n",
" credential=DefaultAzureCredential()\n",
")"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"## What is the A2A Protocol?\n",
"\n",
"The **Agent-to-Agent (A2A) protocol** is an open standard that enables AI agents to communicate,\n",
"discover each other, and collaborate — even when they are built on different frameworks or hosted\n",
"by different services.\n",
"\n",
"Key concepts:\n",
"\n",
"- **Discovery** – Agents publish an *Agent Card* that describes their capabilities, making it\n",
" easy for other agents (or orchestrators) to find the right specialist for a task.\n",
"- **Message Passing** – Agents exchange structured messages through a common protocol, so a\n",
" request from one agent can be understood and fulfilled by another regardless of its internal\n",
" implementation.\n",
"- **Task Lifecycle** – A2A defines states such as *submitted*, *working*, *completed*, and\n",
" *failed*, giving the orchestrator full visibility into how a delegated task is progressing.\n",
"\n",
"In this lesson we simulate A2A-style collaboration by wiring three specialized travel agents\n",
"into a workflow where each agent contributes its expertise and passes results to the next."
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"## Creating Specialized Travel Agents"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"currency_agent = client.as_agent(\n",
" name=\"CurrencyExchangeAgent\",\n",
" instructions=\"\"\"You are a currency exchange specialist. You help travelers understand:\n",
"- Current exchange rates between currencies\n",
"- Best times to exchange money\n",
"- Tips for getting the best rates\n",
"When asked about a destination, provide relevant currency information.\"\"\",\n",
")\n",
"\n",
"activity_agent = client.as_agent(\n",
" name=\"ActivityPlannerAgent\",\n",
" instructions=\"\"\"You are a local activities specialist. You recommend:\n",
"- Must-see attractions and hidden gems\n",
"- Local experiences and cultural activities\n",
"- Restaurant and dining recommendations\n",
"Tailor suggestions to the traveler's interests.\"\"\",\n",
")\n",
"\n",
"travel_manager = client.as_agent(\n",
" name=\"TravelManagerAgent\",\n",
" instructions=\"\"\"You are a travel manager who coordinates between specialist agents.\n",
"When planning a trip:\n",
"1. Gather currency information from the currency specialist\n",
"2. Get activity recommendations from the activity planner\n",
"3. Synthesize everything into a cohesive travel brief\n",
"Present the final plan in an organized, easy-to-read format.\"\"\",\n",
")"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"## Multi-Agent Collaboration via Workflow\n",
"\n",
"We connect the three agents into a sequential workflow that mirrors A2A message passing:\n",
"\n",
"1. **CurrencyExchangeAgent** receives the user request and produces currency guidance.\n",
"2. **ActivityPlannerAgent** receives the enriched context and adds activity recommendations.\n",
"3. **TravelManagerAgent** synthesizes both inputs into a final travel brief."
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"workflow = WorkflowBuilder(start_executor=currency_agent) \\\n",
" .add_edge(currency_agent, activity_agent) \\\n",
" .add_edge(activity_agent, travel_manager) \\\n",
" .build()\n",
"\n",
"last_author = None\n",
"events = workflow.run(\n",
" \"Plan a week-long trip to Tokyo. I love food, temples, and technology.\",\n",
" stream=True,\n",
")\n",
"async for event in events:\n",
" if event.type == \"output\" and isinstance(event.data, AgentResponseUpdate):\n",
" update = event.data\n",
" author = update.author_name\n",
" if author != last_author:\n",
" if last_author is not None:\n",
" print()\n",
" print(f\"\\n{'='*50}\")\n",
" print(f\"🤖 {author}:\")\n",
" print(f\"{'='*50}\")\n",
" last_author = author\n",
" print(update.text, end=\"\", flush=True)"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"## Understanding A2A in Production\n",
"\n",
"In a production environment the A2A protocol unlocks powerful cross-service scenarios:\n",
"\n",
"| Capability | Description |\n",
"|---|---|\n",
"| **Cross-framework interop** | An agent built with one framework can delegate tasks to an agent built with any other A2A-compliant framework, enabling true cross-organization interoperability. |\n",
"| **Service boundaries** | Agents can live in separate microservices, cloud regions, or even different organisations while still collaborating seamlessly. |\n",
"| **Dynamic discovery** | An orchestrator can query an Agent Card registry at runtime to find the best-suited specialist for a given sub-task. |\n",
"| **Streaming & push notifications** | A2A supports Server-Sent Events (SSE) for real-time progress updates and push notifications for long-running tasks. |\n",
"\n",
"The workflow we built above is a simplified, in-process version of this pattern. In a real\n",
"deployment each agent would expose an HTTP endpoint, publish an Agent Card, and communicate\n",
"via the A2A JSON-RPC protocol."
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"## Summary\n",
"\n",
"In this lesson you learned:\n",
"\n",
"1. **What the A2A protocol is** — an open standard for agent-to-agent discovery, messaging,\n",
" and task management.\n",
"2. **How to create specialized agents** — a Currency Exchange agent, an Activity Planner agent,\n",
" and a Travel Manager orchestrator.\n",
"3. **How to wire agents into a workflow** — using `WorkflowBuilder` to model sequential\n",
" message passing between agents.\n",
"4. **How A2A works in production** — enabling cross-framework, cross-service collaboration\n",
" with dynamic discovery and streaming updates."
]
}
],
"metadata": {
"kernelspec": {
"display_name": "Python 3",
"language": "python",
"name": "python3"
},
"language_info": {
"codemirror_mode": {
"name": "ipython",
"version": 3
},
"file_extension": ".py",
"mimetype": "text/x-python",
"name": "python",
"nbconvert_exporter": "python",
"pygments_lexer": "ipython3",
"version": "3.12.13"
}
},
"nbformat": 4,
"nbformat_minor": 2
}
11-agentic-protocols/code_samples/11-mcp-agent-framework.ipynb
+283
View File
@@ -0,0 +1,283 @@
{
"cells": [
{
"cell_type": "markdown",
"metadata": {},
"source": [
"# Lesson 11 - Model Context Protocol (MCP)\n",
"\n",
"The **Model Context Protocol (MCP)** is an open standard that enables agents to dynamically discover and use tools, resources, and data sources at runtime. Instead of hardcoding tools into an agent, MCP lets agents connect to external servers that expose capabilities on demand.\n",
"\n",
"In this lesson, you'll learn:\n",
"- What MCP is and why it matters for agent systems\n",
"- How the MCP client-server architecture works\n",
"- How to build agents that use MCP-style tool discovery"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"## Setup\n",
"\n",
"**Prerequisites:**\n",
"- Microsoft Foundry project with a deployed model\n",
"- Run `az login` for authentication"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"%pip install agent-framework azure-ai-projects azure-identity python-dotenv -q"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"import logging\n",
"logging.getLogger(\"agent_framework.foundry\").setLevel(logging.ERROR)\n",
"\n",
"import os\n",
"import dotenv\n",
"from typing import Annotated\n",
"from agent_framework import tool\n",
"from agent_framework.foundry import FoundryChatClient\n",
"from azure.identity import DefaultAzureCredential\n",
"\n",
"dotenv.load_dotenv()\n",
"\n",
"endpoint = os.getenv(\"AZURE_AI_PROJECT_ENDPOINT\")\n",
"deployment_name = os.getenv(\"AZURE_AI_MODEL_DEPLOYMENT_NAME\")\n",
"\n",
"missing = [k for k, v in {\n",
" \"AZURE_AI_PROJECT_ENDPOINT\": endpoint,\n",
" \"AZURE_AI_MODEL_DEPLOYMENT_NAME\": deployment_name\n",
"}.items() if not v]\n",
"\n",
"if missing:\n",
" raise ValueError(\n",
" f\"Missing required environment variables: {', '.join(missing)}. \"\n",
" \"Please set them as environment variables (e.g., in your .env file or shell environment).\"\n",
" )\n",
"\n",
"# Create the Microsoft Foundry client\n",
"client = FoundryChatClient(\n",
" project_endpoint=endpoint,\n",
" model=deployment_name,\n",
" credential=DefaultAzureCredential()\n",
")"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"## What is the Model Context Protocol (MCP)?\n",
"\n",
"MCP defines a standard way for AI agents to discover and interact with external tools and data sources:\n",
"\n",
"- **MCP Server**: Exposes tools, resources, and prompts via a standard protocol\n",
"- **MCP Client**: The agent runtime that connects to servers and discovers available capabilities\n",
"- **Dynamic Discovery**: Agents don't need hardcoded tools — they discover what's available at runtime\n",
"\n",
"This is powerful for building extensible agent systems where new capabilities can be added without modifying the agent code."
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"## How MCP Works\n",
"\n",
"```\n",
"┌─────────────┐ discover ┌─────────────────┐\n",
"│ MCP Client │ ──────────────► │ MCP Server │\n",
"│ (Agent) │ │ (Tool Provider) │\n",
"│ │ ◄────────────── │ │\n",
"│ │ tool results │ • list_tools() │\n",
"│ │ │ • call_tool() │\n",
"└─────────────┘ │ • resources │\n",
" └─────────────────┘\n",
"```\n",
"\n",
"1. The agent (MCP client) connects to an MCP server\n",
"2. The server responds with a list of available tools and their schemas\n",
"3. The agent can then call any discovered tool during its reasoning\n",
"4. Results flow back through the same protocol"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"## Simulating MCP Tool Discovery\n",
"\n",
"Since a real MCP server requires a running server process, we'll demonstrate the pattern using `@tool` functions that simulate what an MCP-connected accommodation service would provide.\n",
"\n",
"In production, these tools would be discovered dynamically from an MCP server rather than defined locally."
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"@tool(approval_mode=\"never_require\")\n",
"def search_accommodations(\n",
" location: Annotated[str, \"The city to search for accommodations\"],\n",
" check_in: Annotated[str, \"Check-in date (YYYY-MM-DD)\"],\n",
" check_out: Annotated[str, \"Check-out date (YYYY-MM-DD)\"],\n",
" guests: Annotated[int, \"Number of guests\"] = 2\n",
") -> str:\n",
" \"\"\"Search for accommodations (simulating an MCP-connected Airbnb tool).\n",
" In production, this would be discovered via MCP from an accommodation service.\"\"\"\n",
" listings = {\n",
" \"Tokyo\": [\n",
" {\"name\": \"Shinjuku Modern Apartment\", \"price\": 120, \"rating\": 4.8},\n",
" {\"name\": \"Traditional Ryokan in Asakusa\", \"price\": 200, \"rating\": 4.9},\n",
" {\"name\": \"Shibuya Studio\", \"price\": 85, \"rating\": 4.5},\n",
" ],\n",
" \"Paris\": [\n",
" {\"name\": \"Le Marais Charming Flat\", \"price\": 150, \"rating\": 4.7},\n",
" {\"name\": \"Montmartre Artist Loft\", \"price\": 110, \"rating\": 4.6},\n",
" ],\n",
" \"Barcelona\": [\n",
" {\"name\": \"Gothic Quarter Penthouse\", \"price\": 130, \"rating\": 4.8},\n",
" {\"name\": \"Barceloneta Beach Flat\", \"price\": 95, \"rating\": 4.4},\n",
" ],\n",
" }\n",
" results = listings.get(location, [])\n",
" if not results:\n",
" return f\"No accommodations found in {location}\"\n",
" output = f\"Accommodations in {location} ({check_in} to {check_out}, {guests} guests):\\n\"\n",
" for listing in results:\n",
" output += f\" - {listing['name']}: ${listing['price']}/night (★{listing['rating']})\\n\"\n",
" return output\n",
"\n",
"\n",
"@tool(approval_mode=\"never_require\")\n",
"def get_local_experiences(\n",
" location: Annotated[str, \"The city to find experiences in\"],\n",
" interest: Annotated[str, \"Type of experience (food, culture, adventure, etc.)\"] = \"all\"\n",
") -> str:\n",
" \"\"\"Get local experiences and activities (simulating an MCP-connected tourism tool).\"\"\"\n",
" experiences = {\n",
" \"Tokyo\": {\n",
" \"food\": [\"Tsukiji Market Tour ($45)\", \"Ramen Making Class ($60)\", \"Sake Tasting ($35)\"],\n",
" \"culture\": [\"Tea Ceremony ($50)\", \"Samurai Museum ($15)\", \"Sumo Tournament ($80)\"],\n",
" \"adventure\": [\"Mt. Fuji Day Trip ($120)\", \"Go-kart City Tour ($80)\"],\n",
" },\n",
" \"Paris\": {\n",
" \"food\": [\"Wine & Cheese Tasting ($55)\", \"Cooking Class ($90)\", \"Market Tour ($40)\"],\n",
" \"culture\": [\"Louvre Guided Tour ($35)\", \"Montmartre Art Walk ($25)\"],\n",
" },\n",
" }\n",
" city_exp = experiences.get(location, {})\n",
" if not city_exp:\n",
" return f\"No experiences found in {location}\"\n",
" if interest != \"all\" and interest in city_exp:\n",
" items = city_exp[interest]\n",
" return f\"{interest.title()} experiences in {location}:\\n\" + \"\\n\".join(f\" - {e}\" for e in items)\n",
" output = f\"All experiences in {location}:\\n\"\n",
" for cat, items in city_exp.items():\n",
" output += f\"\\n {cat.title()}:\\n\"\n",
" for item in items:\n",
" output += f\" - {item}\\n\"\n",
" return output"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"## Building an Agent with MCP-Style Tools"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"agent = client.as_agent(\n",
" tools=[search_accommodations, get_local_experiences],\n",
" name=\"AccommodationAgent\",\n",
" instructions=\"\"\"You are an accommodation and travel experiences specialist powered by MCP-connected services.\n",
"\n",
"Help travelers find the perfect place to stay and things to do. When searching:\n",
"1. Use the search_accommodations tool to find listings\n",
"2. Use the get_local_experiences tool to suggest activities\n",
"3. Compare options and make personalized recommendations\n",
"4. Consider the traveler's budget, interests, and travel style\"\"\",\n",
")\n",
"\n",
"response = await agent.run(\n",
" \"I'm visiting Tokyo for 5 nights in April with my partner. We love traditional Japanese culture and food. \"\n",
" \"Find us a place to stay and suggest some experiences.\",\n",
" )\n",
"print(response)"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"## MCP in Production\n",
"\n",
"In a production environment, MCP enables powerful patterns:\n",
"\n",
"- **Dynamic tool discovery**: Agents connect to MCP servers and discover tools at runtime\n",
"- **Decoupled architecture**: Tool providers can update independently of the agent\n",
"- **Cross-organization sharing**: Teams can expose capabilities via MCP servers that any agent can use\n",
"- **Microsoft Agent Framework support**: MAF has built-in MCP client support via the `mcp` integration\n",
"\n",
"To use a real MCP server with MAF, you would connect via `hosted_mcp_tool()` or the MCP client integration.\n",
"\n",
"**Learn more:**\n",
"- [MCP Specification](https://modelcontextprotocol.io/)\n",
"- [Microsoft Agent Framework MCP Support](https://github.com/microsoft/agent-framework/tree/main/python/samples/02-agents/mcp)"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"## Summary\n",
"\n",
"In this lesson, you learned:\n",
"- **MCP** is an open standard for dynamic tool discovery between agents and tool providers\n",
"- The **client-server architecture** lets agents discover capabilities at runtime\n",
"- MCP enables **extensible, decoupled agent systems** where tools can be added without code changes\n",
"- Microsoft Agent Framework provides **built-in MCP support** for production use"
]
}
],
"metadata": {
"kernelspec": {
"display_name": "Python 3",
"language": "python",
"name": "python3"
},
"language_info": {
"codemirror_mode": {
"name": "ipython",
"version": 3
},
"file_extension": ".py",
"mimetype": "text/x-python",
"name": "python",
"nbconvert_exporter": "python",
"pygments_lexer": "ipython3",
"version": "3.12.13"
}
},
"nbformat": 4,
"nbformat_minor": 4
}
11-agentic-protocols/code_samples/github-mcp/MCP_SETUP.md
+70
View File
@@ -0,0 +1,70 @@
# MCP Server Integration Guide
## Prerequisites
- Node.js installed (version 14 or higher)
- npm package manager
- Python environment with required dependencies
## Setup Steps
1. **Install MCP Server Package**
```bash
npm install -g @modelcontextprotocol/server-github
```
2. **Start MCP Server**
```bash
npx @modelcontextprotocol/server-github
```
The server should start and display a connection URL.
3. **Verify Connection**
- Look for the plug icon (🔌) in your Chainlit interface
- A number (1) should appear next to the plug icon indicating successful connection
- The console should show: "GitHub plugin setup completed successfully" (along with additional status lines)
## Troubleshooting
### Common Issues
1. **Port Conflict**
```bash
Error: listen EADDRINUSE: address already in use
```
Solution: Change the port using:
```bash
npx @modelcontextprotocol/server-github --port 3001
```
2. **Authentication Issues**
- Ensure GitHub credentials are properly configured
- Check .env file contains required tokens
- Verify GitHub API access
3. **Connection Failed**
- Confirm server is running on expected port
- Check firewall settings
- Verify Python environment has required packages
## Connection Verification
Your MCP server is properly connected when:
1. Console shows "GitHub plugin setup completed successfully"
2. Connection logs show "✓ MCP Connection Status: Active"
3. GitHub commands work in chat interface
## Environment Variables
Required in your .env file:
```
GITHUB_TOKEN=your_github_token
MCP_SERVER_PORT=3000 # Optional, default is 3000
```
## Testing Connection
Send this test message in chat:
```
Show me the repositories for username: [GitHub Username]
```
A successful response will show repository information.
11-agentic-protocols/code_samples/github-mcp/README.md
+62
View File
@@ -0,0 +1,62 @@
# Github MCP Server Example
## Description
This was a demo created for the AI Agents Hackathon hosted through the Microsoft Reactor.
The tools is used to recommend hackathon projects based on a user's Github repos.
This is done by:
1. **Github Agent** - Using the Github MCP Server to retrieve repos and information about those repos.
2. **Hackathon Agent** - Takes the data from the Github Agent and comes up with creative hackathon project ideas based on the projects, languages used by the user and the project tracks for the AI Agents hackathon.
3. **Events Agent** - Based on the hackathon agents suggestion, the events agent will recommend relevant events from the AI Agent Hackathon series.
## Running the code
### Environment Variables
This demo uses Microsoft Agent Framework, Azure OpenAI Service, the Github MCP Server and Azure AI Search.
Make sure that you have the proper environment variables set to use these tools:
```python
AZURE_AI_PROJECT_ENDPOINT=""
AZURE_AI_MODEL_DEPLOYMENT_NAME=""
AZURE_SEARCH_SERVICE_ENDPOINT=""
AZURE_SEARCH_API_KEY=""
```
## Running the Chainlit Server
To connect to the MCP server, this demo use Chainlit as a chat interface.
To run the server, use the following command in your terminal:
```bash
chainlit run app.py -w
```
This should start your Chainlit server on `localhost:8000` as well as populate your Azure AI Search Index with the `event-descriptions.md` content.
## Connecting to the MCP Server
To connect to the Github MCP Server, select the "plug" icon underneath the "Type your message here.." chat box:
![MCP Connect](./images/mcp-chainlit-1.png)
From there you can click on the "Connect an MCP" to add the command to connect to the Github MCP Server:
```bash
npx -y @modelcontextprotocol/server-github --env GITHUB_PERSONAL_ACCESS_TOKEN=[YOUR PERSONAL ACCESS TOKEN]
```
Replace "[YOUR PERSONAL ACCESS TOKEN]" with your actual Personal Access Token.
After connecting, you should see a (1) next to the plug icon to confirm that its connected. If not, try restarting the chainlit server with `chainlit run app.py -w`.
## Using the Demo
To start the agent workflow of recommending hackathon projects, you can type a message like:
"Recommend hackathon projects for the Github user koreyspace"
The Router Agent will analyze your request and determine which combination of agents (GitHub, Hackathon, and Events) is best suited to handle your query. The agents work together to provide comprehensive recommendations based on GitHub repository analysis, project ideation, and relevant tech events.
11-agentic-protocols/code_samples/github-mcp/app.py
+401
View File
@@ -0,0 +1,401 @@
import os
import json
import logging
logging.getLogger("agent_framework.foundry").setLevel(logging.ERROR)
from typing import Annotated
from dotenv import load_dotenv
import requests
import re
import chainlit as cl
from mcp import ClientSession
from agent_framework import tool, AgentResponseUpdate, WorkflowBuilder
from agent_framework.foundry import FoundryChatClient
from azure.identity import AzureCliCredential
from azure.core.credentials import AzureKeyCredential
from azure.search.documents import SearchClient
from azure.search.documents.indexes import SearchIndexClient
from azure.search.documents.indexes.models import SearchIndex, SimpleField, SearchFieldDataType, SearchableField
# Load environment variables
load_dotenv()
# Configure logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)
# Initialize Azure AI Search with persistent storage
search_service_endpoint = os.getenv("AZURE_SEARCH_SERVICE_ENDPOINT")
search_api_key = os.getenv("AZURE_SEARCH_API_KEY")
index_name = "event-descriptions"
search_client = SearchClient(
endpoint=search_service_endpoint,
index_name=index_name,
credential=AzureKeyCredential(search_api_key)
)
index_client = SearchIndexClient(
endpoint=search_service_endpoint,
credential=AzureKeyCredential(search_api_key)
)
# Define the index schema
fields = [
SimpleField(name="id", type=SearchFieldDataType.String, key=True),
SearchableField(name="content", type=SearchFieldDataType.String)
]
index = SearchIndex(name=index_name, fields=fields)
# Check if index already exists if not, create it
try:
existing_index = index_client.get_index(index_name)
print(f"Index '{index_name}' already exists, using the existing index.")
except Exception as e:
# Create the index if it doesn't exist
print(f"Creating new index '{index_name}'...")
index_client.create_index(index)
# Always read event descriptions from markdown file
current_dir = os.path.dirname(os.path.abspath(__file__))
event_descriptions_path = os.path.join(current_dir, "event-descriptions.md")
try:
with open(event_descriptions_path, "r", encoding='utf-8') as f:
markdown_content = f.read()
except FileNotFoundError:
logger.warning(f"Could not find {event_descriptions_path}")
markdown_content = ""
# Split the markdown content into individual event descriptions
event_descriptions = markdown_content.split("---") # You can change the delimiter
# Create documents for Azure Search
documents = []
for i, description in enumerate(event_descriptions):
description = description.strip() # Remove leading/trailing whitespace
if description: # Avoid empty descriptions
documents.append({"id": str(i + 1), "content": description})
# Add documents to the index (only if we have documents)
if documents:
# Delete existing documents first to avoid duplicates
try:
search_client.delete_documents(documents=[{"id": doc["id"]} for doc in documents])
print("Cleared existing documents")
except Exception as e:
print(f"Warning: Failed to clear existing documents: {str(e)}")
# Upload new documents
search_client.upload_documents(documents)
print(f"Uploaded {len(documents)} documents to index")
# RAG tool for event search
@tool
def search_events(
query: Annotated[str, "The search query to find relevant events"]
) -> str:
"""Searches for relevant events based on a query using Azure Search and a live API."""
context_strings = []
try:
results = search_client.search(query, top=5)
for result in results:
if 'content' in result:
context_strings.append(f"Event: {result['content']}")
except Exception as e:
context_strings.append(f"Error searching Azure Search: {str(e)}")
# Live API (example: Devpost hackathons)
try:
api_resp = requests.get(f"https://devpost.com/api/hackathons?search={query}", timeout=5)
if api_resp.ok:
data = api_resp.json()
for event in data.get('hackathons', [])[:5]:
context_strings.append(f"Live Event: {event.get('title')} - {event.get('url')}")
except Exception as e:
context_strings.append(f"Error fetching live events: {str(e)}")
if context_strings:
return "\n\n".join(context_strings)
else:
return "No relevant events found."
def flatten(xss):
return [x for xs in xss for x in xs]
GITHUB_INSTRUCTIONS = """
You are an expert on GitHub repositories. When answering questions, you **must** use the provided GitHub username to find specific information about that user's repositories, including:
* Who created the repositories
* The programming languages used
* Information found in files and README.md files within those repositories
* Provide links to each repository referenfced in your answers
**Important:** Never perform general searches for repositories. Always use the given GitHub username to find the relevant information. If a GitHub username is not provided, state that you need a username to proceed.
"""
HACKATHON_AGENT = """
You are an AI Agent Hackathon Strategist specializing in recommending winning project ideas.
Your task:
1. Analyze the GitHub activity of users to understand their technical skills
2. Suggest creative AI Agent projects tailored to their expertise.
3. Focus on projects that align with Microsoft's AI Agent Hackathon prize categories
When making recommendations:
- Base your ideas strictly on the user's GitHub repositories, languages, and tools
- Give suggestions on tools, languages and frameworks to use to build it.
- Provide detailed project descriptions including architecture and implementation approach
- Explain why the project has potential to win in specific prize categories
- Highlight technical feasibility given the user's demonstrated skills by referencing the specific repositories or languages used.
Formatting your response:
- Provide a clear and structured response that includes:
- Suggested Project Name
- Project Description
- Potential languages and tools to use
- Link to each relevant GitHub repository you based your recommendation on
Hackathon prize categories:
- Best Overall Agent ($20,000)
- Best Agent in Python ($5,000)
- Best Agent in C# ($5,000)
- Best Agent in Java ($5,000)
- Best Agent in JavaScript/TypeScript ($5,000)
- Best Copilot Agent using Microsoft Copilot Studio or Microsoft 365 Agents SDK ($5,000)
- Best Microsoft Foundry Agent Service Usage ($5,000)
"""
EVENTS_AGENT = """
You are an Event Recommendation Agent specializing in suggesting relevant tech events.
Your task:
1. Review the project idea recommended by the Hackathon Agent
2. Use the search_events function to find relevant events based on the technologies mentioned.
3. NEVER suggest and event that the where there is not a relevant technology that the user has used.
3. ONLY recommend events that were returned by the search_events functionf
When making recommendations:
- IMPORTANT: You must first call the search_events function with appropriate technology keywords from the project
- Only recommend events that were explicitly returned by the search_events function
- Do not make up or suggest events that weren't in the search results
- Construct search queries using specific technologies mentioned (e.g., "Python AI workshop" or "JavaScript hackathon")
- Try multiple search queries if needed to find the most relevant events
For each recommended event:
- Only include events found in the search_events results
- Explain the direct connection between the event and the specific project requirements
- Highlight relevant workshops, sessions, or networking opportunities
Formatting your response:
- Start with "Based on the hackathon project idea, here are relevant events that I found:"
- Only list events that were returned by the search_events function
- For each event, include the exact event details as returned by search_events
- Explain specifically how each event relates to the project technologies
If no relevant events are found, acknowledge this and suggest trying different search terms instead of making up events.
"""
@cl.on_mcp_connect
async def on_mcp(connection, session: ClientSession):
logger.info(f"MCP Connection established: {connection.name}")
result = await session.list_tools()
tools = [{
"name": t.name,
"description": t.description,
"input_schema": t.inputSchema,
} for t in result.tools]
mcp_tools = cl.user_session.get("mcp_tools", {})
mcp_tools[connection.name] = tools
cl.user_session.set("mcp_tools", mcp_tools)
# Log available tools
print(f"Available MCP tools for {connection.name}:")
for t in tools:
print(f" - {t['name']}: {t['description']}")
@cl.step(type="tool")
async def call_tool(tool_use):
tool_name = tool_use.name
tool_input = tool_use.input
current_step = cl.context.current_step
current_step.name = tool_name
# Identify which mcp is used
mcp_tools = cl.user_session.get("mcp_tools", {})
mcp_name = None
for connection_name, tools in mcp_tools.items():
if any(t.get("name") == tool_name for t in tools):
mcp_name = connection_name
break
if not mcp_name:
current_step.output = json.dumps(
{"error": f"Tool {tool_name} not found in any MCP connection"})
return current_step.output
mcp_session, _ = cl.context.session.mcp_sessions.get(mcp_name)
if not mcp_session:
current_step.output = json.dumps(
{"error": f"MCP {mcp_name} not found in any MCP connection"})
return current_step.output
try:
current_step.output = await mcp_session.call_tool(tool_name, tool_input)
except Exception as e:
current_step.output = json.dumps({"error": str(e)})
return current_step.output
@cl.on_chat_start
async def on_chat_start():
# Create the Microsoft Foundry Agent Service provider
provider = FoundryChatClient(
project_endpoint=os.environ["AZURE_AI_PROJECT_ENDPOINT"],
model=os.environ["AZURE_AI_MODEL_DEPLOYMENT_NAME"],
credential=AzureCliCredential(),
)
# Create agents using MAF
github_agent = provider.as_agent(
name="GithubAgent",
instructions=GITHUB_INSTRUCTIONS,
)
hackathon_agent = provider.as_agent(
name="HackathonAgent",
instructions=HACKATHON_AGENT,
)
events_agent = provider.as_agent(
name="EventsAgent",
instructions=EVENTS_AGENT,
)
# Build a sequential workflow: GitHub → Hackathon → Events
workflow = WorkflowBuilder(start_executor=github_agent) \
.add_edge(github_agent, hackathon_agent) \
.add_edge(hackathon_agent, events_agent) \
.build()
# Store in user session
cl.user_session.set("provider", provider)
cl.user_session.set("github_agent", github_agent)
cl.user_session.set("hackathon_agent", hackathon_agent)
cl.user_session.set("events_agent", events_agent)
cl.user_session.set("workflow", workflow)
cl.user_session.set("mcp_tools", {})
cl.user_session.set("conversation_history", [])
# Add a cleanup handler for when the session ends
@cl.on_chat_end
async def on_chat_end():
pass
def route_user_input(user_input: str):
"""
Analyze user input and return a list of agent names to invoke.
Returns: list of agent names (e.g., ["GitHubAgent", "HackathonAgent", "EventsAgent"])
"""
user_input_lower = user_input.lower()
agents = []
# Example patterns (expand as needed)
if re.search(r"github|repo|repository|commit|pull request", user_input_lower):
agents.append("GitHubAgent")
if re.search(r"hackathon|project idea|competition|challenge|win", user_input_lower):
agents.append("HackathonAgent")
if re.search(r"event|conference|meetup|workshop|webinar", user_input_lower):
agents.append("EventsAgent")
if not agents:
agents = ["GitHubAgent", "HackathonAgent", "EventsAgent"]
return agents
@cl.on_message
async def on_message(message: cl.Message):
workflow = cl.user_session.get("workflow")
github_agent = cl.user_session.get("github_agent")
hackathon_agent = cl.user_session.get("hackathon_agent")
events_agent = cl.user_session.get("events_agent")
conversation_history = cl.user_session.get("conversation_history", [])
user_input = message.content
agent_names = route_user_input(user_input)
conversation_history.append({"role": "user", "content": user_input})
# If more than one agent is selected, use the workflow
if len(agent_names) > 1:
answer = cl.Message(content="Processing your request using: {}...\n\n".format(", ".join(agent_names)))
await answer.send()
agent_responses = []
try:
events = workflow.run(user_input, stream=True, tools=[search_events])
last_author = None
async for event in events:
if event.type == "output" and isinstance(event.data, AgentResponseUpdate):
update = event.data
author = update.author_name or "Agent"
if author != last_author:
if last_author is not None:
await answer.stream_token("\n\n")
await answer.stream_token(f"**{author}**: ")
last_author = author
if update.text:
await answer.stream_token(update.text)
agent_responses.append(f"**{author}**: {update.text}")
full_response = "".join(agent_responses) if agent_responses else answer.content
conversation_history.append({"role": "assistant", "content": full_response})
cl.user_session.set("conversation_history", conversation_history)
answer.content = full_response
await answer.update()
except Exception as e:
await answer.stream_token(f"\n\n❌ Error: {str(e)}\n\n")
conversation_history.append({"role": "assistant", "content": f"Error: {str(e)}"})
cl.user_session.set("conversation_history", conversation_history)
answer.content += f"\n\n❌ Error: {str(e)}"
await answer.update()
else:
# Single agent: route to the appropriate agent
agent_name = agent_names[0]
agent_map = {
"GitHubAgent": github_agent,
"HackathonAgent": hackathon_agent,
"EventsAgent": events_agent,
}
agent = agent_map.get(agent_name, github_agent)
answer = cl.Message(content=f"Processing your request using {agent_name}...\n\n")
await answer.send()
try:
tools_for_agent = [search_events] if agent_name == "EventsAgent" else []
response = await agent.run(user_input, tools=tools_for_agent)
answer.content = str(response)
conversation_history.append({"role": "assistant", "content": answer.content})
cl.user_session.set("conversation_history", conversation_history)
await answer.update()
except Exception as e:
await answer.stream_token(f"\n\n❌ Error: {str(e)}\n\n")
conversation_history.append({"role": "assistant", "content": f"Error: {str(e)}"})
cl.user_session.set("conversation_history", conversation_history)
answer.content += f"\n\n❌ Error: {str(e)}"
await answer.update()
11-agentic-protocols/code_samples/github-mcp/chainlit.md
+14
View File
@@ -0,0 +1,14 @@
# Welcome to Chainlit! 🚀🤖
Hi there, Developer! 👋 We're excited to have you on board. Chainlit is a powerful tool designed to help you prototype, debug and share applications built on top of LLMs.
## Useful Links 🔗
- **Documentation:** Get started with our comprehensive [Chainlit Documentation](https://docs.chainlit.io) 📚
- **Discord Community:** Join our friendly [Chainlit Discord](https://discord.gg/k73SQ3FyUh) to ask questions, share your projects, and connect with other developers! 💬
We can't wait to see what you create with Chainlit! Happy coding! 💻😊
## Welcome screen
To modify the welcome screen, edit the `chainlit.md` file at the root of your project. If you do not want a welcome screen, just leave this file empty.
11-agentic-protocols/code_samples/github-mcp/event-descriptions.md
+181
View File
@@ -0,0 +1,181 @@
## Event Name: Build your code-first app with Microsoft Foundry Agent Service (EMEA/US offering)
## Description
The Microsoft Foundry Agent Service is a seamless blend of service and SDK that simplifies the development of robust AI-driven solutions. In this session, you'll learn how to build your own code-first AI agent with Azure that can answer questions, perform data analysis, and integrate external data sources. You'll also explore more complex architectures, including multiple agents working together.
## URL
<https://developer.microsoft.com/en-us/reactor/events/25325/>
---
## Event Name: Transforming Business Processes with Multi-Agent AI using Semantic Kernel
## Description
Discover the power of multi-agent AI systems through live demonstrations and hands-on learning with patterns including group-chat, reflection, selector, and swarm. Harness the Semantic Kernel Process Framework to automate and scale critical business processes, from customer support to project management using Python
## URL
<https://developer.microsoft.com/en-us/reactor/events/25313/>
---
## Event Name: Building Agentic Applications with AutoGen v0.4
## Description
Getting started to build agents and multi-agent teams using AutoGen v0.4. We will cover an overview of the new AutoGen v0.4 architecture and walk you through how to build a multi-agent team with a web-based user interface.
## URL
<https://developer.microsoft.com/en-us/reactor/events/25327/>
---
## Event Name: Prototyping AI Agents with GitHub Models
## Description
Thanks to GitHub Models, all you need to build your first AI Agent is a GitHub account! GitHub Models includes powerful models like OpenAI gpt-4o, DeepSeek-R1, Llama-3.1, and many more, ready to try out in the playground or in your code.
In this session, we'll demonstrate how to connect to GitHub Models from Python, and then build agents using popular Python packages like PydanticAI, AutoGen, and Semantic Kernel.
You can follow along live in GitHub Codespaces, or try the examples yourself anytime after the session.
## URL
<https://developer.microsoft.com/en-us/reactor/events/25481/>
---
## Event Name: Building agents with an army of models from the Azure AI model catalog
## Description
The Azure AI model catalog offers a big variety of models, with different skills and capabilities. While using an off the shelf model to get you started, as developers use more sophisticated workflows, they can leverage specialized models to make the job in their framework of choice. In this presentation we go over the model catalog offering, and how you can build agents that sit of top of an army of models - while not costing you a fortune.
## URL
<https://developer.microsoft.com/en-us/reactor/events/25328/>
---
## Event Name: Multi-Agent API with LangGraph and Azure Cosmos DB
## Description
The rise of multi-agent AI applications is transforming how we build intelligent systems - but how do you architect them for real-world scalability and performance? In this session, we’ll take a deep dive into a production-grade multi-agent application built with LangGraph for agent orchestration, FastAPI for an API layer, and Azure Cosmos DB as the backbone for state management, vector storage, and transactional data.
Through a detailed code walkthrough, you’ll see how to design and implement an agent-driven workflow that seamlessly integrates retrieval-augmented generation (RAG), memory persistence, and dynamic state transitions. We’ll cover:
Agent collaboration with LangGraph for structured reasoning
Real-time chat history storage using Azure Cosmos DB - the same database that powers the chat history in ChatGPT, the fastest-growing AI agent application in history
Vector search for knowledge retrieval with Cosmos DB's native embeddings support
FastAPI’s async capabilities to keep interactions responsive and scalable
By the end of this session, you’ll have a clear blueprint for building and deploying your own scalable, cloud-native multi-agent applications that harness the power of modern AI and cloud infrastructure. Whether you're an AI engineer, cloud architect, or Python developer, this talk will equip you with practical insights and battle-tested patterns to build the next generation of AI-powered applications
## URL
<https://developer.microsoft.com/en-us/reactor/events/25314/>
---
## Event Name: Your First AI Agent in JS with Microsoft Foundry Agent Service
## Description
Learn how to build your first AI agent using the JavaScript SDK for Microsoft Foundry Agent Service, a fully managed platform that makes development easy. You’ll see how to set it up, connect tools like Azure AI Search, and deploy a simple question-answering agent. With a live demo, you’ll discover how automatic tool calling and managed state simplify the process. Perfect for beginners, this session gives you practical steps and tips to start your AI agent journey with confidence.
## URL
<https://developer.microsoft.com/en-us/reactor/events/25381/>
---
## Event Name: Prompting is the New Scripting: Meet GenAIScript
## Description
jQuery once made web development easier by abstracting away complexities, allowing developers to focus on building rather than battling browser quirks. Today, AI development faces a similar challenge. New patterns emerge constantly and keeping up can be overwhelming, especially as AI tools—especially agentic ones— become more powerful and complex. What if you could leverage cutting-edge AI capabilities to automate tasks using simple, familiar JavaScript abstractions? Enter GenAIScript—a way to integrate AI into your workflow effortlessly, treating prompts like reusable code snippets. In this talk, we’ll explore how GenAIScript makes AI automation agents feel as intuitive as writing JavaScript, helping you streamline repetitive work without the need for deep AI expertise.
## URL
<https://developer.microsoft.com/en-us/reactor/events/25441/>
---
## Event Name: Knowledge-augmented agents with LlamaIndex.TS
## Description
LlamaIndex is known for making it easy to build Retrieval-Augmented Generation (RAG), but our frameworks also make it easy to build agents and multi-agent systems! In this session we'll introduce Workflows, our basic building block for building agentic systems, and build an agent that uses RAG and other tools.
## URL
<https://developer.microsoft.com/en-us/reactor/events/25440/>
---
## Event Name: AI Agents for Java using Microsoft Foundry and GitHub Copilot
## Description
In this session we’ll show you how to embed advanced AI Agent capabilities into your Java applications using Microsoft Foundry, including setting project goals and experimenting with models and securely deploying production-ready solutions at scale. Along the way, you’ll learn how GitHub Copilot (in IntelliJ, VS Code, and Eclipse) can streamline coding and prompt creation, while best practices in model selection, fine-tuning, and agentic workflows ensure responsible and efficient development. Whether you’re new to AI Agents or looking for advanced agent-building techniques, this session will equip you to deliver next-level experiences with the tooling you already know.
## URL
<https://developer.microsoft.com/en-us/reactor/events/25336/>
---
## Event Name: Building Java AI Agents using LangChain4j and Dynamic Sessions
## Description
Unlock the potential of AI Agents in your Java applications by combining LangChain4j with Azure Container Apps (ACA) dynamic sessions connected to Azure AI services. This session showcases a practical example of building an agent capable of interacting with a remote environment, including file management. Learn how to define custom tools, integrate them into agent workflows, and leverage Azure's scalable infrastructure to deploy intelligent, dynamic solutions.
## URL
<https://developer.microsoft.com/en-us/reactor/events/25337/>
---
## Event Name: Irresponsible AI Agents
## Description
Join us as we explore the potential risks of AI agents and tackle the challenge of embedding trustworthy AI practices into conversational AI platforms! This session dives deep into examples of irresponsible AI agents—showcasing jaw-dropping examples of model failures, adversarial jailbreaks, and other risks that erode trust and compliance.
We'll explore Microsoft’s cutting-edge tools for trustworthy AI, including content filters, red teaming strategies, and evaluations—featuring live demos of AI agents behaving both responsibly and irresponsibly in ways you won’t believe.
🔥 What you’ll walk away with:
✅ How to spot and mitigate AI risks before they can be exploited
✅ How to deploy Azure AI Content Safety to detect and mitigate risky behavior
✅ The secret sauce to making AI agents trustworthy
Get ready for a session packed with hype, high-stakes AI drama, and must-know strategies to keep your AI on the right side of history. Don’t just build AI—build AI that matters!
## URL
<https://developer.microsoft.com/en-us/reactor/events/25388/>
---
## Event Name: Build your code-first app with Microsoft Foundry Agent Service (.NET)
## Description
The Microsoft Foundry Agent Service is a seamless blend of service and SDK that simplifies the development of robust AI-driven solutions. In this session, you'll learn how to build your own code-first AI agent with Azure and C# that can answer questions, perform data analysis, and integrate external data sources. You'll also explore more complex architectures, including multiple agents working together.
## URL
<https://developer.microsoft.com/en-us/reactor/events/25370/>
---
## Event Name: AI Agents + .NET Aspire
## Description
In this session we will share some of the most exciting developments on the .NET platform around Agents. Discover the current status of .NET, including its new features and enhancements. Explore the powerful AI Agent capabilities. And we will do some live coding with Agents and.NET Aspire.
## URL
<https://developer.microsoft.com/en-us/reactor/events/25332/>
---
## Event Name: Semantic Kernel with C# to build multi-agent AI applications powered by Azure Cosmos
## Description
We will walk you through a multi-agent application in C# that is built on top of the Semantic Kernel framework. You will understand the concepts behind agentic applications, understand the implementation details and nuances, and learn how to integrate Azure Cosmos DB as the database for various use-cases.
## URL
<https://developer.microsoft.com/en-us/reactor/events/25455/>
11-agentic-protocols/code_samples/github-mcp/images/mcp-chainlit-1.png
BIN
View File
Binary file not shown.
Side by Side

After

Width:  |  Height:  |  Size: 20 KiB

11-agentic-protocols/code_samples/github-mcp/images/mcp-chainlit-2.png
BIN
View File
Binary file not shown.
Side by Side

After

Width:  |  Height:  |  Size: 91 KiB

11-agentic-protocols/code_samples/mcp-agents/.gitignore
+177
View File
@@ -0,0 +1,177 @@
# Byte-compiled / optimized / DLL files
__pycache__/
*.py[cod]
*$py.class
slides.md
# C extensions
*.so
coding/
# Distribution / packaging
.Python
build/
develop-eggs/
dist/
downloads/
eggs/
.eggs/
lib/
lib64/
parts/
sdist/
var/
wheels/
share/python-wheels/
*.egg-info/
.installed.cfg
*.egg
MANIFEST
# PyInstaller
# Usually these files are written by a python script from a template
# before PyInstaller builds the exe, so as to inject date/other infos into it.
*.manifest
*.spec
# Installer logs
pip-log.txt
pip-delete-this-directory.txt
# Unit test / coverage reports
htmlcov/
.tox/
.nox/
.coverage
.coverage.*
.cache
nosetests.xml
coverage.xml
*.cover
*.py,cover
.hypothesis/
.pytest_cache/
cover/
# Translations
*.mo
*.pot
# Django stuff:
*.log
local_settings.py
db.sqlite3
db.sqlite3-journal
# Flask stuff:
instance/
.webassets-cache
# Scrapy stuff:
.scrapy
# Sphinx documentation
docs/_build/
# PyBuilder
.pybuilder/
target/
# Jupyter Notebook
.ipynb_checkpoints
# IPython
profile_default/
ipython_config.py
# pyenv
# For a library or package, you might want to ignore these files since the code is
# intended to run in multiple environments; otherwise, check them in:
# .python-version
# pipenv
# According to pypa/pipenv#598, it is recommended to include Pipfile.lock in version control.
# However, in case of collaboration, if having platform-specific dependencies or dependencies
# having no cross-platform support, pipenv may install dependencies that don't work, or not
# install all needed dependencies.
#Pipfile.lock
# UV
# Similar to Pipfile.lock, it is generally recommended to include uv.lock in version control.
# This is especially recommended for binary packages to ensure reproducibility, and is more
# commonly ignored for libraries.
#uv.lock
# poetry
# Similar to Pipfile.lock, it is generally recommended to include poetry.lock in version control.
# This is especially recommended for binary packages to ensure reproducibility, and is more
# commonly ignored for libraries.
# https://python-poetry.org/docs/basic-usage/#commit-your-poetrylock-file-to-version-control
#poetry.lock
# pdm
# Similar to Pipfile.lock, it is generally recommended to include pdm.lock in version control.
#pdm.lock
# pdm stores project-wide configurations in .pdm.toml, but it is recommended to not include it
# in version control.
# https://pdm.fming.dev/latest/usage/project/#working-with-version-control
.pdm.toml
.pdm-python
.pdm-build/
# PEP 582; used by e.g. github.com/David-OConnor/pyflow and github.com/pdm-project/pdm
__pypackages__/
# Celery stuff
celerybeat-schedule
celerybeat.pid
# SageMath parsed files
*.sage.py
# Environments
.env
.venv
env/
venv/
ENV/
env.bak/
venv.bak/
# Spyder project settings
.spyderproject
.spyproject
# Rope project settings
.ropeproject
# mkdocs documentation
/site
# mypy
.mypy_cache/
.dmypy.json
dmypy.json
# Pyre type checker
.pyre/
# pytype static type analyzer
.pytype/
# Cython debug symbols
cython_debug/
# PyCharm
# JetBrains specific template is maintained in a separate JetBrains.gitignore that can
# be found at https://github.com/github/gitignore/blob/main/Global/JetBrains.gitignore
# and can be added to the global gitignore or merged into this file. For a more nuclear
# option (not recommended) you can uncomment the following to ignore the entire idea folder.
#.idea/
# Ruff stuff:
.ruff_cache/
# PyPI configuration file
.pypirc
11-agentic-protocols/code_samples/mcp-agents/README.md
+508
View File
@@ -0,0 +1,508 @@
# Building Agent-to-Agent Communication Systems with MCP
> TL;DR - Can You Build Agent2Agent Communication on MCP? Yes!
MCP has evolved significantly beyond its original goal of "providing context to LLMs". With recent enhancements including [resumable streams](https://modelcontextprotocol.io/docs/concepts/transports#resumability-and-redelivery), [elicitation](https://modelcontextprotocol.io/specification/2025-06-18/client/elicitation), [sampling](https://modelcontextprotocol.io/specification/2025-06-18/client/sampling), and notifications ([progress](https://modelcontextprotocol.io/specification/2025-06-18/basic/utilities/progress) and [resources](https://modelcontextprotocol.io/specification/2025-06-18/schema#resourceupdatednotification)), MCP now provides a robust foundation for building complex agent-to-agent communication systems.
## The Agent/Tool Misconception
As more developers explore tools with agentic behaviors (run for long periods, may require additional input mid-execution, etc.), a common misconception is that MCP is unsuitable primarily because early examples of its tools primitive focused on simple request-response patterns.
This perception is outdated. The MCP specification has been significantly enhanced over the past few months with capabilities that close the gap for building long-running agentic behavior:
- **Streaming & Partial Results**: Real-time progress updates during execution
- **Resumability**: Clients can reconnect and continue after disconnection
- **Durability**: Results survive server restarts (e.g., via resource links)
- **Multi-turn**: Interactive input mid-execution via elicitation and sampling
These features can be composed to enable complex agentic and multi-agent applications, all deployed on the MCP protocol.
For reference, we will refer to an agent as a "tool" that is available on an MCP server. This implies the existence of a host application which implements an MCP client that establishes a session with the MCP server and can call the agent.
## What Makes an MCP Tool "Agentic"?
Before diving into implementation, let's establish what infrastructure capabilities are needed to support long-running agents.
> We will define an agent as an entity that can operate autonomously over extended periods, capable of handling complex tasks that may require multiple interactions or adjustments based on real-time feedback.
### 1. Streaming & Partial Results
Traditional request-response patterns don't work for long-running tasks. Agents need to provide:
- Real-time progress updates
- Intermediate results
**MCP Support**: Resource update notifications enable streaming partial results, though this requires careful design to avoid conflicts with JSON-RPC's 1:1 request/response model.
| Feature | Use Case | MCP Support |
| -------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ | ------------------------------------------------------------------------------------------ |
| Real-time Progress Updates | User requests a codebase migration task. The agent streams progress: "10% - Analyzing dependencies... 25% - Converting TypeScript files... 50% - Updating imports..." | ✅ Progress notifications |
| Partial Results | "Generate a book" task streams partial results, e.g., 1) Story arc outline, 2) Chapter list, 3) Each chapter as completed. Host can inspect, cancel, or redirect at any stage. | ✅ Notifications can be "extended" to include partial results see proposals on PR 383, 776 |
<div align="center" style="font-style: italic; font-size: 0.95em; margin-bottom: 0.5em;">
<strong>Figure 1:</strong> This diagram illustrates how an MCP agent streams real-time progress updates and partial results to the host application during a long-running task, enabling the user to monitor execution in real time.
</div>
```mermaid
sequenceDiagram
participant User
participant Host as Host App<br/>(MCP Client)
participant Server as MCP Server<br/>(Agent Tool)
User->>Host: Start long task
Host->>Server: Call agent_tool()
loop Progress Updates
Server-->>Host: Progress + partial results
Host-->>User: Stream updates
end
Server-->>Host: ✅ Final result
Host-->>User: Complete
```
### 2. Resumability
Agents must handle network interruptions gracefully:
- Reconnect after (client) disconnection
- Continue from where they left off (message redelivery)
**MCP Support**: MCP StreamableHTTP transport today supports session resumption and message redelivery with session IDs and last event IDs. The important note here is that the server must implement an EventStore that enables event replays on client reconnection.
Note that there is a community proposal (PR #975) that explores transport-agnostic resumable streams.
| Feature | Use Case | MCP Support |
| ------------ | ---------------------------------------------------------------------------------------------------------------------------------------------------------- | -------------------------------------------------------------------------- |
| Resumability | Client disconnects during long-running task. Upon reconnection, session resumes with missed events replayed, continuing seamlessly from where it left off. | ✅ StreamableHTTP transport with session IDs, event replay, and EventStore |
<div align="center" style="font-style: italic; font-size: 0.95em; margin-bottom: 0.5em;">
<strong>Figure 2:</strong> This diagram shows how MCP's StreamableHTTP transport and event store enable seamless session resumption: if the client disconnects, it can reconnect and replay missed events, continuing the task without loss of progress.
</div>
```mermaid
sequenceDiagram
participant User
participant Host as Host App<br/>(MCP Client)
participant Server as MCP Server<br/>(Agent Tool)
participant Store as Event Store
User->>Host: Start task
Host->>Server: Call tool [session: abc123]
Server->>Store: Save events
Note over Host,Server: 💥 Connection lost
Host->>Server: Reconnect [session: abc123]
Store-->>Server: Replay events
Server-->>Host: Catch up + continue
Host-->>User: ✅ Complete
```
### 3. Durability
Long-running agents need persistent state:
- Results survive server restarts
- Status can be retrieved out-of-band
- Progress tracking across sessions
**MCP Support**: MCP now supports a Resource link return type for tool calls. Today, a possible pattern is to design a tool that creates a resource and immediately returns a resource link. The tool can continue to address the task in the background and update the resource. In turn, the client can choose to poll the state of this resource to get partial or full results (based on what resource updates the server provides) or subscribe to the resource for update notifications.
One limitation here is that polling resources or subscribing for updates can consume resources with implications at scale. There is an open community proposal (including #992) exploring the possibility of including webhooks or triggers that the server can call to notify the client/host application of updates.
| Feature | Use Case | MCP Support |
| ---------- | ----------------------------------------------------------------------------------------------------------------------------------------------- | ------------------------------------------------------------------ |
| Durability | Server crashes during data migration task. Results and progress survive restart, client can check status and continue from persistent resource. | ✅ Resource links with persistent storage and status notifications |
Today, a common pattern is to design a tool that creates a resource and immediately returns a resource link. The tool can in the background address the task, issue resource notifications that serve as progress updates or include partial results, and update the content in the resource as needed.
<div align="center" style="font-style: italic; font-size: 0.95em; margin-bottom: 0.5em;">
<strong>Figure 3:</strong> This diagram demonstrates how MCP agents use persistent resources and status notifications to ensure that long-running tasks survive server restarts, allowing clients to check progress and retrieve results even after failures.
</div>
```mermaid
sequenceDiagram
participant User
participant Host as Host App<br/>(MCP Client)
participant Server as MCP Server<br/>(Agent Tool)
participant DB as Persistent Storage
User->>Host: Start task
Host->>Server: Call tool
Server->>DB: Create resource + updates
Server-->>Host: 🔗 Resource link
Note over Server: 💥 Server restart
User->>Host: Check status
Host->>Server: Get resource
Server->>DB: Load state
Server-->>Host: Current progress
Server->>DB: Complete + notify
Host-->>User: ✅ Complete
```
### 4. Multi-Turn Interactions
Agents often need additional input mid-execution:
- Human clarification or approval
- AI assistance for complex decisions
- Dynamic parameter adjustment
**MCP Support**: Fully supported via sampling (for AI input) and elicitation (for human input).
| Feature | Use Case | MCP Support |
| ----------------------- | -------------------------------------------------------------------------------------------------------------------------------------------- | ----------------------------------------------------- |
| Multi-Turn Interactions | Travel booking agent requests price confirmation from user, then asks AI to summarize travel data before completing the booking transaction. | ✅ Elicitation for human input, sampling for AI input |
<div align="center" style="font-style: italic; font-size: 0.95em; margin-bottom: 0.5em;">
<strong>Figure 4:</strong> This diagram depicts how MCP agents can interactively elicit human input or request AI assistance mid-execution, supporting complex, multi-turn workflows such as confirmations and dynamic decision-making.
</div>
```mermaid
sequenceDiagram
participant User
participant Host as Host App<br/>(MCP Client)
participant Server as MCP Server<br/>(Agent Tool)
User->>Host: Book flight
Host->>Server: Call travel_agent
Server->>Host: Elicitation: "Confirm $500?"
Note over Host: Elicitation callback (if available)
Host->>User: 💰 Confirm price?
User->>Host: "Yes"
Host->>Server: Confirmed
Server->>Host: Sampling: "Summarize data"
Note over Host: AI callback (if available)
Host->>Server: Report summary
Server->>Host: ✅ Flight booked
```
## Implementing Long-Running Agents on MCP - Code Overview
As part of this article, we provide a [code repository](https://github.com/victordibia/ai-tutorials/tree/main/MCP%20Agents) that contains a complete implementation of long-running agents using the MCP Python SDK with StreamableHTTP transport for session resumption and message redelivery. The implementation demonstrates how MCP capabilities can be composed to enable sophisticated agent-like behaviors.
Specifically, we implement a server with two primary agent tools:
- **Travel Agent** - Simulates a travel booking service with price confirmation via elicitation
- **Research Agent** - Performs research tasks with AI-assisted summaries via sampling
Both agents demonstrate real-time progress updates, interactive confirmations, and full session resumption capabilities.
### Key Implementation Concepts
The following sections show server-side agent implementation and client-side host handling for each capability:
#### Streaming & Progress Updates - Real-time Task Status
Streaming enables agents to provide real-time progress updates during long-running tasks, keeping users informed of task status and intermediate results.
**Server Implementation (agent sends progress notifications):**
```python
# From server/server.py - Travel agent sending progress updates
for i, step in enumerate(steps):
await ctx.session.send_progress_notification(
progress_token=ctx.request_id,
progress=i * 25,
total=100,
message=step,
related_request_id=str(ctx.request_id)
)
await anyio.sleep(2) # Simulate work
# Alternative: Log messages for detailed step-by-step updates
await ctx.session.send_log_message(
level="info",
data=f"Processing step {current_step}/{steps} ({progress_percent}%)",
logger="long_running_agent",
related_request_id=ctx.request_id,
)
```
**Client Implementation (host receives progress updates):**
```python
# From client/client.py - Client handling real-time notifications
async def message_handler(message) -> None:
if isinstance(message, types.ServerNotification):
if isinstance(message.root, types.LoggingMessageNotification):
console.print(f"📡 [dim]{message.root.params.data}[/dim]")
elif isinstance(message.root, types.ProgressNotification):
progress = message.root.params
console.print(f"🔄 [yellow]{progress.message} ({progress.progress}/{progress.total})[/yellow]")
# Register message handler when creating session
async with ClientSession(
read_stream, write_stream,
message_handler=message_handler
) as session:
```
#### Elicitation - Requesting User Input
Elicitation enables agents to request user input mid-execution. This is essential for confirmations, clarifications, or approvals during long-running tasks.
**Server Implementation (agent requests confirmation):**
```python
# From server/server.py - Travel agent requesting price confirmation
elicit_result = await ctx.session.elicit(
message=f"Please confirm the estimated price of $1200 for your trip to {destination}",
requestedSchema=PriceConfirmationSchema.model_json_schema(),
related_request_id=ctx.request_id,
)
if elicit_result and elicit_result.action == "accept":
# Continue with booking
logger.info(f"User confirmed price: {elicit_result.content}")
elif elicit_result and elicit_result.action == "decline":
# Cancel the booking
booking_cancelled = True
```
**Client Implementation (host provides elicitation callback):**
```python
# From client/client.py - Client handling elicitation requests
async def elicitation_callback(context, params):
console.print(f"💬 Server is asking for confirmation:")
console.print(f" {params.message}")
response = console.input("Do you accept? (y/n): ").strip().lower()
if response in ['y', 'yes']:
return types.ElicitResult(
action="accept",
content={"confirm": True, "notes": "Confirmed by user"}
)
else:
return types.ElicitResult(
action="decline",
content={"confirm": False, "notes": "Declined by user"}
)
# Register the callback when creating the session
async with ClientSession(
read_stream, write_stream,
elicitation_callback=elicitation_callback
) as session:
```
#### Sampling - Requesting AI Assistance
Sampling allows agents to request LLM assistance for complex decisions or content generation during execution. This enables hybrid human-AI workflows.
**Server Implementation (agent requests AI assistance):**
```python
# From server/server.py - Research agent requesting AI summary
sampling_result = await ctx.session.create_message(
messages=[
SamplingMessage(
role="user",
content=TextContent(type="text", text=f"Please summarize the key findings for research on: {topic}")
)
],
max_tokens=100,
related_request_id=ctx.request_id,
)
if sampling_result and sampling_result.content:
if sampling_result.content.type == "text":
sampling_summary = sampling_result.content.text
logger.info(f"Received sampling summary: {sampling_summary}")
```
**Client Implementation (host provides sampling callback):**
```python
# From client/client.py - Client handling sampling requests
async def sampling_callback(context, params):
message_text = params.messages[0].content.text if params.messages else 'No message'
console.print(f"🧠 Server requested sampling: {message_text}")
# In a real application, this could call an LLM API
# For demo purposes, we provide a mock response
mock_response = "Based on current research, MCP has evolved significantly..."
return types.CreateMessageResult(
role="assistant",
content=types.TextContent(type="text", text=mock_response),
model="interactive-client",
stopReason="endTurn"
)
# Register the callback when creating the session
async with ClientSession(
read_stream, write_stream,
sampling_callback=sampling_callback,
elicitation_callback=elicitation_callback
) as session:
```
#### Resumability - Session Continuity Across Disconnections
Resumability ensures that long-running agent tasks can survive client disconnections and continue seamlessly upon reconnection. This is implemented through event stores and resumption tokens.
**Event Store Implementation (server holds session state):**
```python
# From server/event_store.py - Simple in-memory event store
class SimpleEventStore(EventStore):
def __init__(self):
self._events: list[tuple[StreamId, EventId, JSONRPCMessage]] = []
self._event_id_counter = 0
async def store_event(self, stream_id: StreamId, message: JSONRPCMessage) -> EventId:
"""Store an event and return its ID."""
self._event_id_counter += 1
event_id = str(self._event_id_counter)
self._events.append((stream_id, event_id, message))
return event_id
async def replay_events_after(self, last_event_id: EventId, send_callback: EventCallback) -> StreamId | None:
"""Replay events after the specified ID for resumption."""
# Find events after the last known event and replay them
for _, event_id, message in self._events[start_index:]:
await send_callback(EventMessage(message, event_id))
# From server/server.py - Passing event store to session manager
def create_server_app(event_store: Optional[EventStore] = None) -> Starlette:
server = ResumableServer()
# Create session manager with event store for resumption
session_manager = StreamableHTTPSessionManager(
app=server,
event_store=event_store, # Event store enables session resumption
json_response=False,
security_settings=security_settings,
)
return Starlette(routes=[Mount("/mcp", app=session_manager.handle_request)])
# Usage: Initialize with event store
event_store = SimpleEventStore()
app = create_server_app(event_store)
```
**Client Metadata with Resumption Token (client reconnects using stored state):**
```python
# From client/client.py - Client resumption with metadata
if existing_tokens and existing_tokens.get("resumption_token"):
# Use existing resumption token to continue where we left off
metadata = ClientMessageMetadata(
resumption_token=existing_tokens["resumption_token"],
)
else:
# Create callback to save resumption token when received
def enhanced_callback(token: str):
protocol_version = getattr(session, 'protocol_version', None)
token_manager.save_tokens(session_id, token, protocol_version, command, args)
metadata = ClientMessageMetadata(
on_resumption_token_update=enhanced_callback,
)
# Send request with resumption metadata
result = await session.send_request(
types.ClientRequest(
types.CallToolRequest(
method="tools/call",
params=types.CallToolRequestParams(name=command, arguments=args)
)
),
types.CallToolResult,
metadata=metadata,
)
```
The host application maintains session IDs and resumption tokens locally, enabling it to reconnect to existing sessions without losing progress or state.
### Code Organization
<div align="center" style="font-style: italic; font-size: 0.95em; margin-bottom: 0.5em;">
<strong>Figure 5:</strong> MCP-based agent system architecture
</div>
```mermaid
graph LR
User([User]) -->|"Task"| Host["Host<br/>(MCP Client)"]
Host -->|list tools| Server[MCP Server]
Server -->|Exposes| AgentsTools[Agents as Tools]
AgentsTools -->|Task| AgentA[Travel Agent]
AgentsTools -->|Task| AgentB[Research Agent]
Host -->|Monitors| StateUpdates[Progress & State Updates]
Server -->|Publishes| StateUpdates
class User user;
class AgentA,AgentB agent;
class Host,Server,StateUpdates core;
```
**Key Files:**
- **`server/server.py`** - Resumable MCP server with travel and research agents that demonstrate elicitation, sampling, and progress updates
- **`client/client.py`** - Interactive host application with resumption support, callback handlers, and token management
- **`server/event_store.py`** - Event store implementation enabling session resumption and message redelivery
## Extending to Multi-Agent Communication on MCP
The implementation above can be extended to multi-agent systems by enhancing the host application's intelligence and scope:
- **Intelligent Task Decomposition**: Host analyzes complex user requests and breaks them into subtasks for different specialized agents
- **Multi-Server Coordination**: Host maintains connections to multiple MCP servers, each exposing different agent capabilities
- **Task State Management**: Host tracks progress across multiple concurrent agent tasks, handling dependencies and sequencing
- **Resilience & Retries**: Host manages failures, implements retry logic, and reroutes tasks when agents become unavailable
- **Result Synthesis**: Host combines outputs from multiple agents into coherent final results
The host evolves from a simple client to an intelligent orchestrator, coordinating distributed agent capabilities while maintaining the same MCP protocol foundation.
## Conclusion
MCP's enhanced capabilities - resource notifications, elicitation/sampling, resumable streams, and persistent resources - enable complex agent-to-agent interactions while maintaining protocol simplicity.
## Getting Started
Ready to build your own agent2agent system? Follow these steps:
### 1. Run the Demo
```bash
# Start the server with event store for resumption
python -m server.server --port 8006
# In another terminal, run the interactive client
python -m client.client --url http://127.0.0.1:8006/mcp
```
**Available commands in interactive mode:**
- `travel_agent` - Book travel with price confirmation via elicitation
- `research_agent` - Research topics with AI-assisted summaries via sampling
- `list` - Show all available tools
- `clean-tokens` - Clear resumption tokens
- `help` - Show detailed command help
- `quit` - Exit the client
### 2. Test Resumption Capabilities
- Start a long-running agent (e.g., `travel_agent`)
- Interrupt the client during execution (Ctrl+C)
- Restart the client - it will automatically resume from where it left off
### 3. Explore and Extend
- **Explore the examples**: Check out this [mcp-agents](https://github.com/victordibia/ai-tutorials/tree/main/MCP%20Agents)
- **Join the community**: Participate in MCP discussions on GitHub
- **Experiment**: Start with a simple long-running task and gradually add streaming, resumability, and multi-agent coordination
This demonstrates how MCP enables intelligent agent behaviors while maintaining tool-based simplicity.
Overall, the MCP protocol spec is evolving rapidly; the reader is encouraged to review the official documentation website for the most recent updates - https://modelcontextprotocol.io/introduction
11-agentic-protocols/code_samples/mcp-agents/client/__init__.py
+8
View File
@@ -0,0 +1,8 @@
"""
MCP Client implementations.
This package contains various MCP client implementations:
- client.py: Basic MCP client
- resumable_client.py: Client with resumption support
- utils.py: Utility functions for token management, etc.
"""
11-agentic-protocols/code_samples/mcp-agents/client/client.py
+456
View File
@@ -0,0 +1,456 @@
#!/usr/bin/env python3
"""
Interactive MCP Client for Agent-to-Agent Communication Tutorial
"""
import asyncio
import argparse
import logging
from typing import Dict, Any
from mcp import ClientSession
from mcp.client.streamable_http import streamablehttp_client
from mcp.server.streamable_http import MCP_SESSION_ID_HEADER, MCP_PROTOCOL_VERSION_HEADER
from mcp.shared.message import ClientMessageMetadata
import mcp.types as types
from rich.console import Console
from rich.panel import Panel
from .utils import TokenManager, cast_input_value
console = Console()
def display_tools(tools):
"""Display available tools as a simple list."""
console.print("[bold]Available Tools:[/bold]")
for tool in tools:
tool_type = "🤖" if any(word in tool.name.lower() for word in ["agent", "travel", "research"]) else "🔧"
console.print(f" {tool_type} [cyan]{tool.name}[/cyan]")
console.print()
console.print("[dim]Type a tool name to run it with default arguments[/dim]")
def extract_text_content(result) -> str:
"""Extract text content from tool result."""
for content in result.content:
if hasattr(content, 'text'):
return content.text
return "No text content available"
async def execute_tool_with_resumption(session, command: str, args: dict, get_session_id, on_resumption_token_update, existing_tokens=None, token_manager=None):
"""Execute a tool with resumption support using send_request."""
current_session_id = get_session_id()
if not current_session_id:
raise RuntimeError("No session ID available - resumption requires a valid session")
session_id = current_session_id
# If we have an existing resumption token, pass it for resumption
if existing_tokens and existing_tokens.get("resumption_token"):
metadata = ClientMessageMetadata(
resumption_token=existing_tokens["resumption_token"],
)
else:
# Create enhanced callback that saves tool context immediately when token is received
def enhanced_callback(token: str):
# Since callback fires immediately with the actual resumption token,
# save everything needed for resumption right away
protocol_version = getattr(session, 'protocol_version', None)
if token_manager:
token_manager.save_tokens(session_id, token, protocol_version, command, args)
# Also call the original callback
return on_resumption_token_update(session_id, token, command, args)
metadata = ClientMessageMetadata(
on_resumption_token_update=enhanced_callback,
)
result = await session.send_request(
types.ClientRequest(
types.CallToolRequest(
method="tools/call",
params=types.CallToolRequestParams(
name=command,
arguments=args
),
)
),
types.CallToolResult,
metadata=metadata,
)
return result
async def interactive_mode(server_url: str):
"""Run interactive mode with tool exploration."""
# Configure logging to suppress noisy SSE parsing errors
logging.getLogger('mcp.client.streamable_http').setLevel(logging.ERROR)
# Filter out specific SSE JSON parsing errors
class SSEFilter(logging.Filter):
def filter(self, record):
# Suppress "Error parsing SSE message" and JSON validation errors
if ("Error parsing SSE message" in record.getMessage() or
"ValidationError" in record.getMessage() or
"EOF while parsing" in record.getMessage()):
return False
return True
# Apply filter to relevant loggers
for logger_name in ['mcp.client.streamable_http', 'mcp', 'pydantic_core']:
logger = logging.getLogger(logger_name)
logger.addFilter(SSEFilter())
console.print(Panel("[bold cyan]🎮 Interactive MCP Client[/bold cyan]", expand=False))
console.print("[dim]Explore MCP server tools interactively[/dim]\n")
# Check for existing resumption tokens
token_manager = TokenManager()
existing_tokens: Dict[str, Any] | None = token_manager.load_tokens()
# Prepare headers for resumption if tokens exist
headers = {}
if existing_tokens:
headers[MCP_SESSION_ID_HEADER] = existing_tokens["session_id"]
if "protocol_version" in existing_tokens:
headers[MCP_PROTOCOL_VERSION_HEADER] = existing_tokens["protocol_version"]
console.print(f"[cyan]🔄 Found existing session, attempting resumption...[/cyan]")
try:
async with streamablehttp_client(
server_url,
headers=headers if headers else None,
terminate_on_close=False # Enable resumption
) as (read_stream, write_stream, get_session_id):
# Create message handler for real-time notifications
async def message_handler(message) -> None:
try:
if isinstance(message, types.ServerNotification):
if isinstance(message.root, types.LoggingMessageNotification):
console.print(f"📡 [dim]{message.root.params.data}[/dim]")
elif isinstance(message.root, types.ProgressNotification):
progress = message.root.params
console.print(f"🔄 [yellow]{progress.message} ({progress.progress}/{progress.total})[/yellow]")
elif isinstance(message.root, types.ResourceUpdatedNotification):
console.print(f"� [blue]Resource updated: {message.root.params.uri}[/blue]")
except Exception:
# Silently ignore message handler errors
pass
# Add sampling callback for research agent
async def sampling_callback(context, params):
try:
message_text = params.messages[0].content.text if params.messages else 'No message'
console.print(f"\n🧠 [bold cyan]Server requested sampling:[/bold cyan]")
console.print(f" [yellow]{message_text}[/yellow]")
# Mock response instead of prompting user
mock_response = "Based on current research, MCP has evolved significantly with new features like resumable streams, elicitation, and sampling capabilities, enabling sophisticated agent-to-agent communication patterns."
console.print(f"[dim]🤖 Auto-responding with mock data:[/dim]")
console.print(f" [green]{mock_response[:80]}...[/green]")
return types.CreateMessageResult(
role="assistant",
content=types.TextContent(
type="text",
text=mock_response
),
model="interactive-client",
stopReason="endTurn"
)
except Exception as e:
return types.ErrorData(code=-1, message=str(e))
# Add elicitation callback for travel agent
async def elicitation_callback(context, params):
try:
console.print(f"\n💬 [bold yellow]Server is asking for confirmation:[/bold yellow]")
console.print(f" [cyan]{params.message}[/cyan]")
# Get user's decision
while True:
response = console.input("\n[bold]Do you accept? (y/n/details): [/bold]").strip().lower()
if response in ['y', 'yes', 'accept']:
user_notes = console.input("[dim]Any additional notes (optional): [/dim]").strip()
console.print(f"[dim]✅ Sending acceptance with notes: '{user_notes or 'Confirmed by user'}'[/dim]")
return types.ElicitResult(
action="accept",
content={
"confirm": True,
"notes": user_notes if user_notes else "Confirmed by user"
}
)
elif response in ['n', 'no', 'decline']:
reason = console.input("[dim]Reason for declining (optional): [/dim]").strip()
return types.ElicitResult(
action="decline",
content={
"confirm": False,
"notes": reason if reason else "Declined by user"
}
)
elif response in ['d', 'details']:
console.print(f"[dim]Schema: {params.requestedSchema if hasattr(params, 'requestedSchema') else 'Not specified'}[/dim]")
else:
console.print("[red]Please enter 'y' (yes), 'n' (no), or 'd' (details)[/red]")
except Exception as e:
console.print(f"[red]Error in elicitation: {e}[/red]")
return types.ElicitResult(action="decline", content={"confirm": False, "notes": f"Error: {e}"})
# Add resumption token callback for long-running tools
async def on_resumption_token_update(session_id: str, resumption_token: str, tool_name: str, tool_args: dict):
"""Callback for when resumption token is updated during long-running operations."""
# Save the updated token with tool information
protocol_version = existing_tokens.get("protocol_version") if existing_tokens else None
token_manager.save_tokens(session_id, resumption_token, protocol_version, tool_name, tool_args)
async with ClientSession(
read_stream,
write_stream,
message_handler=message_handler,
sampling_callback=sampling_callback,
elicitation_callback=elicitation_callback
) as session:
# Handle resumption vs new session based on tokens
if existing_tokens and existing_tokens.get("resumption_token") and existing_tokens.get("last_tool") and existing_tokens.get("last_args"):
# RESUMPTION FLOW: We have everything needed for resumption
console.print(f"[cyan]🔄 Resuming existing session...[/cyan]")
console.print(f"[cyan] Session ID: {existing_tokens['session_id']}[/cyan]")
console.print(f"[cyan] Resuming tool: {existing_tokens['last_tool']}[/cyan]")
console.print(f"[green]✅ Skipping initialization, directly resuming...[/green]")
try:
# Directly resume the tool execution with cached data
result = await execute_tool_with_resumption(
session,
existing_tokens["last_tool"],
existing_tokens["last_args"],
get_session_id,
on_resumption_token_update,
existing_tokens, # Pass existing tokens for resumption
token_manager
)
console.print(f"[green]✅ Resumed tool completed successfully![/green]")
console.print(f"[green]Result:[/green]")
console.print(f" {extract_text_content(result)}")
# Task completed successfully, clean up tokens
if token_manager.delete_tokens():
console.print(f"[dim]🧹 Cleaned up resumption tokens[/dim]")
# After successful resumption, continue with normal interactive flow
console.print("\n[cyan]📋 Loading available tools for continued interaction...[/cyan]")
tools_result = await session.list_tools()
tools = tools_result.tools
tools_dict = {tool.name: tool for tool in tools}
display_tools(tools)
except Exception as resume_error:
error_msg = str(resume_error)
if "ValidationError" in error_msg and "JSON" in error_msg:
console.print(f"[yellow]⚠️ Resumed tool completed with connection issues, but likely succeeded[/yellow]")
# Even if there was a connection issue, task likely completed
if token_manager.delete_tokens():
console.print(f"[dim]🧹 Cleaned up resumption tokens[/dim]")
# Load tools for continued interaction
console.print("\n[cyan]📋 Loading available tools for continued interaction...[/cyan]")
tools_result = await session.list_tools()
tools = tools_result.tools
tools_dict = {tool.name: tool for tool in tools}
display_tools(tools)
else:
console.print(f"[red]❌ Resume failed: {resume_error}[/red]")
console.print(f"[yellow]Falling back to normal initialization...[/yellow]")
# Fallback to normal initialization
result = await session.initialize()
console.print(f"[green]✅ Connected to: {result.serverInfo.name}[/green]")
tools_result = await session.list_tools()
tools = tools_result.tools
tools_dict = {tool.name: tool for tool in tools}
display_tools(tools)
else:
# NORMAL FLOW: No resumption tokens, do full initialization
console.print("[cyan]🔌 Connecting to server...[/cyan]")
result = await session.initialize()
console.print(f"[green]✅ Connected to: {result.serverInfo.name}[/green]")
console.print(f"[green]📋 Protocol: {result.protocolVersion}[/green]")
# Save protocol version for future resumption
current_session_id = get_session_id()
if current_session_id:
console.print(f"[green]🆔 Session: {current_session_id}[/green]")
# Load and display available tools
console.print("\n[cyan]📋 Loading available tools...[/cyan]")
tools_result = await session.list_tools()
tools = tools_result.tools
tools_dict = {tool.name: tool for tool in tools}
display_tools(tools)
# Interactive command loop
console.print("\n[dim]Commands: [tool_name], 'list', 'help', 'clean-tokens', 'quit'[/dim]")
while True:
try:
command = console.input("\n[bold blue]> [/bold blue]").strip()
if command.lower() in ['quit', 'exit', 'q']:
console.print("[yellow]👋 Goodbye![/yellow]")
break
elif command.lower() in ['list', 'l']:
display_tools(tools)
elif command.lower() in ['help', 'h']:
console.print("[bold]Available Commands:[/bold]")
console.print(" [cyan]list[/cyan] or [cyan]l[/cyan] - Show available tools")
console.print(" [cyan][tool_name][/cyan] - Execute a tool")
console.print(" [cyan]help[/cyan] or [cyan]h[/cyan] - Show this help")
console.print(" [cyan]clean-tokens[/cyan] - Delete resumption tokens")
console.print(" [cyan]quit[/cyan] or [cyan]q[/cyan] - Exit")
console.print("\n[bold]Agent tools (try these!):[/bold]")
for tool in tools:
if "agent" in tool.name.lower():
console.print(f" 🤖 [yellow]{tool.name}[/yellow]")
elif command.lower() in ['clean-tokens', 'clean']:
if token_manager.delete_tokens():
console.print("[green]✅ Resumption tokens cleared[/green]")
else:
console.print("[yellow]No tokens to clear[/yellow]")
elif command in tools_dict:
tool = tools_dict[command]
# Collect arguments
args = {}
if tool.inputSchema and tool.inputSchema.get("properties"):
console.print(f"[dim]Tool '{command}' parameters:[/dim]")
for prop_name, prop_info in tool.inputSchema["properties"].items():
required = prop_name in tool.inputSchema.get("required", [])
default = prop_info.get("default", "")
desc = prop_info.get("description", "")
prompt = f" {prop_name}"
if desc:
prompt += f" ({desc})"
if default:
prompt += f" [default: {default}]"
if required:
prompt += " [required]"
prompt += ": "
value = console.input(prompt).strip()
if value:
# Cast the input value to the correct type
args[prop_name] = cast_input_value(value, prop_info)
elif required and not default:
console.print(f"[red]❌ {prop_name} is required[/red]")
break
else:
# All required args collected, execute tool
console.print(f"[cyan]🔧 Executing {command}...[/cyan]")
try:
result = await execute_tool_with_resumption(
session, command, args, get_session_id, on_resumption_token_update, existing_tokens, token_manager
)
console.print(f"[green]✅ Result:[/green]")
console.print(f" {extract_text_content(result)}")
# Task completed successfully, clean up tokens
if token_manager.delete_tokens():
console.print(f"[dim]🧹 Cleaned up resumption tokens[/dim]")
except Exception as tool_error:
error_msg = str(tool_error)
if "ValidationError" in error_msg and "JSON" in error_msg:
console.print(f"[yellow]⚠️ Tool completed with connection issues, but likely succeeded[/yellow]")
# Even if there was a connection issue, task likely completed
if token_manager.delete_tokens():
console.print(f"[dim]🧹 Cleaned up resumption tokens[/dim]")
else:
console.print(f"[red]❌ Tool failed: {tool_error}[/red]")
else:
# No parameters needed, execute tool
console.print(f"[cyan]🔧 Executing {command}...[/cyan]")
try:
result = await execute_tool_with_resumption(
session, command, {}, get_session_id, on_resumption_token_update, existing_tokens, token_manager
)
console.print(f"[green]✅ Result:[/green]")
console.print(f" {extract_text_content(result)}")
# Task completed successfully, clean up tokens
if token_manager.delete_tokens():
console.print(f"[dim]🧹 Cleaned up resumption tokens[/dim]")
except Exception as tool_error:
error_msg = str(tool_error)
if "ValidationError" in error_msg and "JSON" in error_msg:
console.print(f"[yellow]⚠️ Tool completed with connection issues, but likely succeeded[/yellow]")
# Even if there was a connection issue, task likely completed
if token_manager.delete_tokens():
console.print(f"[dim]🧹 Cleaned up resumption tokens[/dim]")
else:
console.print(f"[red]❌ Tool failed: {tool_error}[/red]")
else:
console.print(f"[red]❌ Unknown command: '{command}'[/red]")
console.print("[dim]Type 'list' to see available tools or 'help' for commands[/dim]")
except KeyboardInterrupt:
console.print("\n[yellow]👋 Interrupted![/yellow]")
break
except Exception as e:
console.print(f"[red]❌ Connection error: {e}[/red]")
async def main():
"""Main entry point."""
parser = argparse.ArgumentParser(description="Interactive MCP Client")
parser.add_argument("--url", default="http://127.0.0.1:8006/mcp", help="MCP server URL")
parser.add_argument("--clean-tokens", action="store_true", help="Delete existing resumption tokens and start fresh")
args = parser.parse_args()
# Handle token cleaning
if args.clean_tokens:
token_manager = TokenManager()
if token_manager.delete_tokens():
console.print("[green]✅ Resumption tokens cleared[/green]")
else:
console.print("[yellow]No tokens to clear[/yellow]")
return
await interactive_mode(args.url)
if __name__ == "__main__":
# Configure logging to suppress noisy MCP warnings
logging.basicConfig(level=logging.WARNING)
# Suppress specific noisy loggers
logging.getLogger('mcp.client.streamable_http').setLevel(logging.ERROR)
logging.getLogger('httpx').setLevel(logging.WARNING)
try:
asyncio.run(main())
except KeyboardInterrupt:
console.print("\n[yellow]👋 Goodbye![/yellow]")
except Exception as e:
console.print(f"[red]💥 Fatal error: {e}[/red]")
11-agentic-protocols/code_samples/mcp-agents/client/resumable_client.py
+224
View File
@@ -0,0 +1,224 @@
#!/usr/bin/env python3
"""
Minimal Resumable MCP Client
Simple client that demonstrates session resumption with the long_running_agent tool.
- If resumption token exists, resumes the session
- If no token exists, creates new session and saves token
- Clears token when task completes
"""
import argparse
import asyncio
import json
import logging
import os
from pathlib import Path
from typing import Dict, Any, Optional
from mcp import ClientSession, types
from mcp.client.streamable_http import streamablehttp_client
from mcp.server.streamable_http import MCP_SESSION_ID_HEADER, MCP_PROTOCOL_VERSION_HEADER
from mcp.shared.message import ClientMessageMetadata
from mcp.types import TextContent
# Configure logging to suppress noisy SSE warnings
logging.basicConfig(level=logging.WARNING)
logger = logging.getLogger(__name__)
# Suppress specific noisy loggers
logging.getLogger('httpx').setLevel(logging.WARNING)
logging.getLogger('mcp.client.streamable_http').setLevel(logging.ERROR)
logging.getLogger('mcp').setLevel(logging.WARNING)
logging.getLogger('pydantic_core').setLevel(logging.ERROR)
# Filter out specific SSE JSON parsing errors
class SSEFilter(logging.Filter):
def filter(self, record):
return not (
"Error parsing SSE message" in record.getMessage() or
"Invalid JSON: EOF while parsing" in record.getMessage()
)
# Apply filter to relevant loggers
for logger_name in ['mcp.client.streamable_http', 'mcp', 'pydantic_core']:
log = logging.getLogger(logger_name)
log.addFilter(SSEFilter())
# Token file location
TOKEN_FILE = Path(".mcp_resumption_token.json")
def load_resumption_tokens() -> Optional[Dict[str, Any]]:
"""Load resumption tokens from file."""
if TOKEN_FILE.exists():
try:
with open(TOKEN_FILE, 'r') as f:
tokens = json.load(f)
logger.info(f"🔄 Loaded resumption tokens: session_id={tokens.get('session_id')}")
return tokens
except Exception as e:
logger.error(f"Error loading tokens: {e}")
return None
def save_resumption_tokens(session_id: str, resumption_token: str, protocol_version: str = "2025-03-26"):
"""Save resumption tokens to file."""
tokens = {
"session_id": session_id,
"resumption_token": resumption_token,
"protocol_version": protocol_version,
"tool_name": "long_running_agent",
"tool_args": {}
}
try:
with open(TOKEN_FILE, 'w') as f:
json.dump(tokens, f, indent=2)
logger.info(f"💾 Saved resumption tokens: session_id={session_id}")
except Exception as e:
logger.error(f"Error saving tokens: {e}")
def clear_resumption_tokens():
"""Clear resumption tokens file."""
if TOKEN_FILE.exists():
try:
TOKEN_FILE.unlink()
logger.info("🗑️ Cleared resumption tokens")
except Exception as e:
logger.error(f"Error clearing tokens: {e}")
async def run_long_running_task(server_url: str):
"""Run the long_running_agent with resumption support."""
# Check for existing tokens
existing_tokens = load_resumption_tokens()
# Prepare headers for resumption if tokens exist
headers = {}
if existing_tokens:
headers[MCP_SESSION_ID_HEADER] = existing_tokens["session_id"]
headers[MCP_PROTOCOL_VERSION_HEADER] = existing_tokens.get("protocol_version", "2025-03-26")
print("🔄 Resuming existing session...")
else:
print("🆕 Creating new session...")
try:
async with streamablehttp_client(
server_url,
headers=headers if headers else None,
terminate_on_close=False # Enable resumption
) as (read_stream, write_stream, get_session_id):
# Message handler for real-time notifications (standardized to use log messages)
async def message_handler(message) -> None:
try:
if isinstance(message, types.ServerNotification):
if isinstance(message.root, types.LoggingMessageNotification):
log_data = message.root.params
# Format log messages similar to progress notifications for consistency
print(f"📡 [{log_data.logger}] {log_data.data}")
elif isinstance(message.root, types.ProgressNotification):
# Keep support for progress notifications in case other tools use them
progress = message.root.params
print(f"📊 Progress: {progress.progress}/{progress.total} - {progress.message}")
elif isinstance(message.root, types.ResourceUpdatedNotification):
print(f"🔄 Resource updated: {message.root.params.uri}")
except Exception:
# Silently ignore message handler errors to avoid breaking the flow
pass
async with ClientSession(read_stream, write_stream, message_handler=message_handler) as session:
# Only initialize if this is a new session (no existing tokens)
if not existing_tokens:
result = await session.initialize()
print(f"✅ Session initialized: {result.serverInfo.name}")
else:
print("✅ Using existing session (no initialization needed)")
# Set up resumption token callback (matching test signature)
async def on_resumption_token_update(token: str) -> None:
print(f"💾 Resumption token received: {token[:20]}...")
session_id = get_session_id()
if session_id:
protocol_version = getattr(session, 'negotiated_protocol_version', '2025-03-26')
save_resumption_tokens(session_id, token, protocol_version)
print(f"💾 Resumption token saved")
# Prepare metadata with resumption support
if existing_tokens and existing_tokens.get("resumption_token"):
# Resume existing task
print("🔄 Resuming long-running task...")
metadata = ClientMessageMetadata(
resumption_token=existing_tokens["resumption_token"]
)
else:
# Start new task
print("🚀 Starting long-running task...")
metadata = ClientMessageMetadata(
on_resumption_token_update=on_resumption_token_update
)
# Execute the long_running_agent tool
try:
print("metadata:", metadata)
result = await session.send_request(
types.ClientRequest(
types.CallToolRequest(
method="tools/call",
params=types.CallToolRequestParams(
name="long_running_agent",
arguments={}
),
)
),
types.CallToolResult,
metadata=metadata,
)
# Task completed successfully
content_text = "Task completed"
if result.content and len(result.content) > 0:
first_content = result.content[0]
if isinstance(first_content, TextContent):
content_text = first_content.text
print(f"✅ Task completed: {content_text}")
clear_resumption_tokens()
except Exception as e:
print(f"❌ Task failed: {e}")
# Keep tokens in case we want to retry
except Exception as e:
print(f"❌ Connection error: {e}")
async def main():
"""Main entry point."""
parser = argparse.ArgumentParser(description="Minimal Resumable MCP Client")
parser.add_argument("--url", default="http://127.0.0.1:8006/mcp", help="MCP server URL")
parser.add_argument("--clear-tokens", action="store_true", help="Clear resumption tokens and exit")
args = parser.parse_args()
if args.clear_tokens:
clear_resumption_tokens()
print("🗑️ Resumption tokens cleared")
return
print("🤖 Minimal Resumable MCP Client")
print(f"🔗 Connecting to: {args.url}")
await run_long_running_task(args.url)
if __name__ == "__main__":
try:
asyncio.run(main())
except KeyboardInterrupt:
print("\n👋 Interrupted by user")
except Exception as e:
print(f"💥 Fatal error: {e}")
11-agentic-protocols/code_samples/mcp-agents/client/utils.py
+119
View File
@@ -0,0 +1,119 @@
#!/usr/bin/env python3
"""
Client Utilities for MCP Session Management
This module provides utilities for managing MCP client sessions,
including token persistence and session resumption.
"""
import json
import os
from typing import Optional, Dict, Any
from rich.console import Console
# Import shared constants locally
DEFAULT_TOKEN_FILE = "resumption_tokens.json"
console = Console()
def cast_input_value(value: str, prop_info: dict):
"""Cast input value to the correct type based on schema property info."""
if not value:
return value
prop_type = prop_info.get("type", "string")
try:
if prop_type == "integer":
return int(value)
elif prop_type == "number":
return float(value)
elif prop_type == "boolean":
return value.lower() in ['true', 't', 'yes', 'y', '1']
else: # string or other types
return value
except ValueError:
# If casting fails, return the original string
return value
class TokenManager:
"""Manages resumption tokens and session information."""
def __init__(self, token_file: str = DEFAULT_TOKEN_FILE):
self.token_file = token_file
def save_tokens(self, session_id: str, resumption_token: str, protocol_version: Optional[str] = None,
last_tool: Optional[str] = None, last_args: Optional[dict] = None) -> bool:
"""Save session ID, resumption token, protocol version, and last tool info to file.
Only creates a token file when we have an actual resumption token from the server.
Never creates empty or placeholder token files.
"""
# Validate that we have real tokens, not placeholders
if not session_id or not resumption_token:
console.print(f"[red]✗ Cannot save tokens: missing session_id or resumption_token[/red]")
return False
if resumption_token.startswith("pending_") or resumption_token.startswith("temp_"):
console.print(f"[red]✗ Cannot save placeholder token: {resumption_token}[/red]")
return False
data: Dict[str, Any] = {
"session_id": session_id,
"resumption_token": resumption_token,
}
if protocol_version:
data["protocol_version"] = protocol_version
if last_tool:
data["last_tool"] = last_tool
if last_args:
data["last_args"] = last_args
try:
with open(self.token_file, 'w') as f:
json.dump(data, f, indent=2)
console.print(f"[green]✓ Saved resumption token to {self.token_file}[/green]")
return True
except Exception as e:
console.print(f"[red]✗ Failed to save tokens: {e}[/red]")
return False
def load_tokens(self) -> Optional[Dict[str, Any]]:
"""Load session ID, resumption token, and tool info from file."""
if not os.path.exists(self.token_file):
console.print(f"[yellow]No resumption token file found ({self.token_file})[/yellow]")
return None
try:
with open(self.token_file, 'r') as f:
data = json.load(f)
if "session_id" in data and "resumption_token" in data:
console.print(f"[cyan]📄 Found resumption tokens in {self.token_file}[/cyan]")
console.print(f"[cyan] Session ID: {data['session_id']}[/cyan]")
console.print(f"[cyan] Token: {data['resumption_token'][:20]}...[/cyan]")
if "last_tool" in data:
console.print(f"[cyan] Last Tool: {data['last_tool']}[/cyan]")
return data
else:
console.print(f"[yellow]Invalid token file format[/yellow]")
return None
except Exception as e:
console.print(f"[red]✗ Failed to load tokens: {e}[/red]")
return None
def delete_tokens(self) -> bool:
"""Delete the resumption token file."""
try:
if os.path.exists(self.token_file):
os.remove(self.token_file)
return True
else:
return False
except Exception as e:
console.print(f"[red]✗ Failed to delete token file: {e}[/red]")
return False
11-agentic-protocols/code_samples/mcp-agents/server/__init__.py
+8
View File
@@ -0,0 +1,8 @@
"""
MCP Server implementations.
This package contains various MCP server implementations:
- server.py: Basic MCP server
- resumable_server.py: Server with event store and resumption support
- event_store.py: Event store implementation for session resumption
"""
11-agentic-protocols/code_samples/mcp-agents/server/event_store.py
+242
View File
@@ -0,0 +1,242 @@
#!/usr/bin/env python3
"""
Event Store Implementation for MCP Session Resumption
This module provides event store implementations that enable MCP session resumption
by storing and replaying events after client reconnection.
"""
import asyncio
import logging
import sqlite3
from typing import Optional
from pydantic import TypeAdapter
from mcp.server.streamable_http import (
EventCallback,
EventId,
EventMessage,
EventStore,
StreamId,
)
from mcp.types import JSONRPCMessage
logger = logging.getLogger(__name__)
class SimpleEventStore(EventStore):
"""Simple in-memory event store for testing resumption functionality."""
def __init__(self):
self._events: list[tuple[StreamId, EventId, JSONRPCMessage]] = []
self._event_id_counter = 0
logger.info("SimpleEventStore initialized")
async def store_event(self, stream_id: StreamId, message: JSONRPCMessage) -> EventId:
"""Store an event and return its ID."""
self._event_id_counter += 1
event_id = str(self._event_id_counter)
self._events.append((stream_id, event_id, message))
logger.info(f"Stored event {event_id} for stream {stream_id}")
return event_id
async def replay_events_after(
self,
last_event_id: EventId,
send_callback: EventCallback,
) -> StreamId | None:
"""Replay events after the specified ID."""
logger.info(f"Replaying events after {last_event_id}")
# Find the index of the last event ID
start_index = None
for i, (_, event_id, _) in enumerate(self._events):
if event_id == last_event_id:
start_index = i + 1
break
if start_index is None:
# If event ID not found, start from beginning
start_index = 0
logger.info("Event ID not found, starting from beginning")
stream_id = None
# Replay events
replayed_count = 0
for _, event_id, message in self._events[start_index:]:
await send_callback(EventMessage(message, event_id))
replayed_count += 1
# Capture the stream ID from the first replayed event
if stream_id is None and len(self._events) > start_index:
stream_id = self._events[start_index][0]
logger.info(f"Replayed {replayed_count} events, stream_id: {stream_id}")
return stream_id
def get_event_count(self) -> int:
"""Get the total number of stored events."""
return len(self._events)
def clear_events(self) -> None:
"""Clear all stored events."""
self._events.clear()
self._event_id_counter = 0
logger.info("Event store cleared")
class PersistentEventStore(EventStore):
"""
Event store that persists events to disk using SQLite.
"""
def __init__(self, storage_path: str = "events.db") -> None:
self.storage_path = storage_path
self._adapter = TypeAdapter(JSONRPCMessage)
# Use check_same_thread=False to allow access from asyncio executor threads
self._conn = sqlite3.connect(self.storage_path, check_same_thread=False)
self._create_table()
logger.info(f"PersistentEventStore initialized with {self.storage_path}")
def _create_table(self) -> None:
"""Create the events table if it doesn't exist."""
cursor = self._conn.cursor()
try:
cursor.execute("""
CREATE TABLE IF NOT EXISTS events (
id INTEGER PRIMARY KEY AUTOINCREMENT,
stream_id TEXT NOT NULL,
message TEXT NOT NULL
)
""")
self._conn.commit()
except sqlite3.Error:
logger.exception("Failed to create 'events' table in PersistentEventStore")
try:
self._conn.close()
except Exception:
logger.exception("Failed to close SQLite connection after table creation error")
raise
finally:
try:
cursor.close()
except Exception:
logger.exception("Failed to close SQLite cursor after table creation")
async def store_event(self, stream_id: StreamId, message: JSONRPCMessage) -> EventId:
"""Store an event and return its ID."""
# Serialize message to JSON
json_str = self._adapter.dump_json(message).decode('utf-8')
# Run DB operation in thread pool to avoid blocking event loop
return await asyncio.to_thread(self._store_event_sync, stream_id, json_str)
def _store_event_sync(self, stream_id: StreamId, json_str: str) -> EventId:
cursor = self._conn.cursor()
cursor.execute(
"INSERT INTO events (stream_id, message) VALUES (?, ?)",
(stream_id, json_str)
)
self._conn.commit()
event_id = str(cursor.lastrowid)
logger.info(f"Stored event {event_id} for stream {stream_id}")
return event_id
async def replay_events_after(
self,
last_event_id: EventId,
send_callback: EventCallback,
) -> StreamId | None:
"""Replay events after the specified ID, filtered by the stream of the last event."""
logger.info(f"Replaying events after {last_event_id}")
# Fetch events in thread pool
events_data = await asyncio.to_thread(self._fetch_events_sync, last_event_id)
if events_data is None:
logger.warning(f"Could not resume stream from event {last_event_id}")
return None
stream_id = None
replayed_count = 0
for event_id, row_stream_id, message_json in events_data:
if stream_id is None:
stream_id = row_stream_id
try:
message = self._adapter.validate_json(message_json)
await send_callback(EventMessage(message, event_id))
replayed_count += 1
except Exception as e:
logger.error(f"Failed to deserialize event {event_id}: {e}")
logger.info(f"Replayed {replayed_count} events for stream {stream_id}")
return stream_id
def _fetch_events_sync(self, last_event_id: EventId) -> list[tuple[EventId, StreamId, str]] | None:
try:
target_id = int(last_event_id)
except (ValueError, TypeError):
logger.warning(f"Invalid event ID format: {last_event_id}")
return None
cursor = self._conn.cursor()
# 1. Identify the stream from the last event ID
cursor.execute("SELECT stream_id FROM events WHERE id = ?", (target_id,))
result = cursor.fetchone()
if not result:
logger.warning(f"Event ID {target_id} not found")
return None
stream_id = result[0]
# 2. Fetch subsequent events for THIS STREAM ONLY
cursor.execute(
"SELECT id, stream_id, message FROM events WHERE id > ? AND stream_id = ? ORDER BY id ASC",
(target_id, stream_id)
)
# Convert rows to list of (str_id, stream_id, msg_json)
return [(str(row[0]), row[1], row[2]) for row in cursor.fetchall()]
def get_event_count(self) -> int:
"""Get the total number of stored events."""
cursor = self._conn.cursor()
cursor.execute("SELECT COUNT(*) FROM events")
result = cursor.fetchone()
return result[0] if result else 0
def clear_events(self) -> None:
"""Clear all stored events."""
cursor = self._conn.cursor()
cursor.execute("DELETE FROM events")
# Reset auto-increment sequence
cursor.execute("DELETE FROM sqlite_sequence WHERE name='events'")
self._conn.commit()
logger.info("Event store cleared")
def close(self) -> None:
"""Close the underlying SQLite connection."""
conn = getattr(self, "_conn", None)
if conn is None:
return
try:
conn.close()
logger.info("PersistentEventStore connection closed")
except sqlite3.Error as exc:
logger.warning("Error closing PersistentEventStore connection: %s", exc)
finally:
self._conn = None
def __enter__(self) -> "PersistentEventStore":
"""Enter the runtime context related to this object."""
return self
def __exit__(self, exc_type, exc_val, exc_tb) -> None:
"""Exit the runtime context and close the connection."""
self.close()
11-agentic-protocols/code_samples/mcp-agents/server/server.py
+382
View File
@@ -0,0 +1,382 @@
#!/usr/bin/env python3
"""
Resumable MCP Server Implementation
This server provides full session resumption capabilities using an event store.
It supports long-running tasks that can be resumed after client disconnection.
"""
import argparse
import asyncio
import logging
import re
from typing import Optional
import anyio
import uvicorn
from pydantic import AnyUrl
from starlette.applications import Starlette
from starlette.routing import Mount
from pydantic import BaseModel, Field
from mcp.server import Server
from mcp.server.streamable_http import EventStore
from mcp.server.streamable_http_manager import StreamableHTTPSessionManager
from mcp.server.transport_security import TransportSecuritySettings
from mcp.types import TextContent, Tool, SamplingMessage
from .event_store import SimpleEventStore
logger = logging.getLogger(__name__)
class PriceConfirmationSchema(BaseModel):
confirm: bool = Field(description="Confirm the price for this trip")
notes: str = Field(default="", description="Any additional notes about the price")
class ResumableServer(Server):
"""Server implementation with long-running tools and notifications for resumption testing."""
def __init__(self, name: str = "resumable_mcp_server"):
super().__init__(name)
logger.info(f"ResumableServer '{name}' initialized")
@self.list_tools()
async def handle_list_tools() -> list[Tool]:
"""List available tools including resumable ones."""
return [
Tool(
name="travel_agent",
description="Book a travel trip with progress updates and price confirmation",
inputSchema={
"type": "object",
"properties": {
"destination": {
"type": "string",
"description": "Travel destination",
"default": "Paris"
}
}
},
),
Tool(
name="research_agent",
description="Research a topic with progress updates and interactive summaries",
inputSchema={
"type": "object",
"properties": {
"topic": {
"type": "string",
"description": "Research topic",
"default": "AI trends"
}
}
},
),
Tool(
name="long_running_agent",
description="A long-running task for testing resumption (50 steps, 2 seconds each)",
inputSchema={
"type": "object",
"properties": {}
},
),
]
@self.call_tool()
async def handle_call_tool(name: str, args: dict) -> list[TextContent]:
"""Handle tool execution with support for long-running tasks."""
ctx = self.request_context
logger.info(f"Tool called: {name} with args: {args}")
if name == "travel_agent":
destination = args.get("destination", "Paris")
logger.info(f"Travel agent: destination={destination}")
# Simple travel booking flow with progress updates
steps = [
"Checking flights...",
"Finding available dates...",
"Confirming prices...",
"Booking flight..."
]
elicitation_result = None
booking_cancelled = False
for i, step in enumerate(steps):
await ctx.session.send_progress_notification(
progress_token=ctx.request_id,
progress=i * 25,
total=100,
message=step,
related_request_id=str(ctx.request_id)
)
# Add elicitation request at step 3 (Confirming prices)
if i == 2: # "Confirming prices..." step
try:
elicit_result = await ctx.session.elicit(
message=f"Please confirm the estimated price of $1200 for your trip to {destination}",
requestedSchema=PriceConfirmationSchema.model_json_schema(),
related_request_id=ctx.request_id,
)
elicitation_result = elicit_result
if elicit_result and elicit_result.action == "accept":
logger.info(f"User confirmed price: {elicit_result.content}")
# Continue with booking
elif elicit_result and elicit_result.action == "decline":
logger.info(f"User declined price confirmation: {elicit_result.content}")
booking_cancelled = True
# Stop the booking process
await ctx.session.send_progress_notification(
progress_token=ctx.request_id,
progress=100,
total=100,
message="Booking cancelled by user",
related_request_id= str(ctx.request_id)
)
break
else:
logger.info("User cancelled elicitation")
booking_cancelled = True
await ctx.session.send_progress_notification(
progress_token=ctx.request_id,
progress=100,
total=100,
message="Booking cancelled"
)
break
except Exception as e:
logger.info(f"Elicitation request failed (this is normal in tests): {e}")
# Continue with booking anyway for fallback
if not booking_cancelled:
await anyio.sleep(2) # Fixed 0.5 second delay between steps
# Generate final result based on elicitation outcome
if booking_cancelled:
if elicitation_result and hasattr(elicitation_result, 'content') and elicitation_result.content:
notes = elicitation_result.content.get('notes', 'No reason provided')
result_text = f"❌ Booking cancelled for trip to {destination}. Reason: {notes}"
else:
result_text = f"❌ Booking cancelled for trip to {destination}."
else:
# Final progress update for successful booking
await ctx.session.send_progress_notification(
progress_token=ctx.request_id,
progress=100,
total=100,
message="Trip booked successfully"
)
# Include confirmation details in success message
if elicitation_result and elicitation_result.action == "accept" and elicitation_result.content:
notes = elicitation_result.content.get('notes', 'No additional notes')
result_text = f"✅ Trip booked successfully to {destination}! Price confirmed with notes: '{notes}'"
else:
result_text = f"✅ Trip booked successfully to {destination}!"
return [TextContent(type="text", text=result_text)]
elif name == "research_agent":
topic = args.get("topic", "AI trends")
logger.info(f"Research agent: topic={topic}")
# Simple research flow with progress updates
steps = [
"Gathering sources...",
"Analyzing data...",
"Summarizing findings...",
"Finalizing report..."
]
sampling_summary = None
for i, step in enumerate(steps):
await ctx.session.send_progress_notification(
progress_token=ctx.request_id,
progress=i * 25,
total=100,
message=step
)
# Add sampling request at step 3 (Summarizing findings)
if i == 2: # "Summarizing findings..." step
try:
sampling_result = await ctx.session.create_message(
messages=[
SamplingMessage(
role="user",
content=TextContent(type="text", text=f"Please summarize the key findings for research on: {topic}")
)
],
max_tokens=100,
related_request_id=ctx.request_id,
)
if sampling_result and sampling_result.content:
if sampling_result.content.type == "text":
sampling_summary = sampling_result.content.text
logger.info(f"Received sampling summary: {sampling_summary}")
except Exception as e:
logger.info(f"Sampling request failed (this is normal in tests): {e}")
await anyio.sleep(2) # Fixed 0.5 second delay between steps
# Final progress update
await ctx.session.send_progress_notification(
progress_token=ctx.request_id,
progress=100,
total=100,
message="Research completed successfully"
)
# Use sampling summary if available, otherwise default message
if sampling_summary:
result_text = f"🔍 Research on '{topic}' completed successfully!\n\n📊 Key Findings (from user input): {sampling_summary}"
else:
result_text = f"🔍 Research on '{topic}' completed successfully!"
return [TextContent(type="text", text=result_text)]
elif name == "long_running_agent":
# Fixed values optimized for resumption testing
steps = 50
duration = 2.0
logger.info(f"Long running agent: {steps} steps, {duration}s each")
# Send initial log message
await ctx.session.send_log_message(
level="info",
data="Long-running task started",
logger="long_running_agent",
related_request_id=ctx.request_id,
)
# Execute the long-running task
for i in range(steps):
current_step = i + 1
# Use integer arithmetic to avoid floating point precision issues
progress_percent = (current_step * 100) // steps
# Send log message for each step
await ctx.session.send_log_message(
level="info",
data=f"Processing step {current_step}/{steps} ({progress_percent}%)",
logger="long_running_agent",
related_request_id=ctx.request_id,
)
# Wait for 2 seconds
await anyio.sleep(duration)
# Send completion log message
await ctx.session.send_log_message(
level="info",
data=f"Task completed successfully! Processed {steps} steps in {steps * duration:.0f} seconds.",
logger="long_running_agent",
related_request_id=ctx.request_id,
)
# Final completion message
result_text = f"✅ Long-running task completed successfully! Processed {steps} steps in {steps * duration:.0f} seconds."
return [TextContent(type="text", text=result_text)]
else:
raise ValueError(f"Unknown tool: {name}")
def create_server_app(event_store: Optional[EventStore] = None) -> Starlette:
"""Create the Starlette application with resumable MCP server."""
# Create server instance
server = ResumableServer()
# Create security settings
security_settings = TransportSecuritySettings(
allowed_hosts=["127.0.0.1:*", "localhost:*"],
allowed_origins=["http://127.0.0.1:*", "http://localhost:*"]
)
# Create session manager with event store
session_manager = StreamableHTTPSessionManager(
app=server,
event_store=event_store,
json_response=False, # Use SSE streams
security_settings=security_settings,
)
# Create ASGI application
app = Starlette(
debug=True,
routes=[
Mount("/mcp", app=session_manager.handle_request),
],
lifespan=lambda app: session_manager.run(),
)
return app
async def run_server(port: int = 8006, with_event_store: bool = True) -> None:
"""Run the resumable HTTP server."""
# Create event store if requested
event_store = SimpleEventStore() if with_event_store else None
# Create application
app = create_server_app(event_store)
# Configure server
config = uvicorn.Config(
app=app,
host="127.0.0.1",
port=port,
log_level="info",
limit_concurrency=10,
timeout_keep_alive=30,
access_log=True,
)
logger.info(f"Starting Resumable HTTP MCP Server on http://127.0.0.1:{port}/mcp")
if event_store:
logger.info("Event store enabled - resumption supported")
else:
logger.info("Event store disabled - no resumption support")
# Start the server
server = uvicorn.Server(config=config)
try:
await server.serve()
except KeyboardInterrupt:
logger.info("Server stopped by user")
except Exception as e:
logger.error(f"Server error: {e}")
raise
def main():
"""Main entry point."""
parser = argparse.ArgumentParser(description="Resumable HTTP MCP Server")
parser.add_argument("--port", type=int, default=8006, help="Port to listen on (default: 8006)")
parser.add_argument("--no-event-store", action="store_true", help="Disable event store (no resumption)")
args = parser.parse_args()
# Configure logging
logging.basicConfig(level=logging.INFO)
# Run the server
asyncio.run(run_server(
port=args.port,
with_event_store=not args.no_event_store
))
if __name__ == "__main__":
main()