"""
Product Recommendation MCP Client using Microsoft Agent Framework
Tests all MCP methods using ChatAgent and MCPStreamableHTTPTool
"""
import asyncio
import os
from dotenv import load_dotenv
from agent_framework import ChatAgent, MCPStreamableHTTPTool
from agent_framework.azure import AzureOpenAIChatClient
# Load environment variables
load_dotenv()
# MCP Server configuration
MCP_SERVER_URL = "http://localhost:8000/mcp"
# Global agent instance
agent = None
mcp_tool = None
def print_section(title: str):
"""Print a formatted section header"""
print("\n" + "="*80)
print(f" {title}")
print("="*80)
def print_result(query: str, result: str):
"""Print query and result in a formatted way"""
print(f"\n📝 Query: {query}")
print(f"✅ Result:\n{result}\n")
async def initialize_agent():
"""Initialize the Microsoft Agent Framework ChatAgent with MCP tools"""
global agent, mcp_tool
print_section("INITIALIZING MICROSOFT AGENT FRAMEWORK")
print(f"\n🔗 Connecting to MCP Server: {MCP_SERVER_URL}")
# Initialize MCP Server Tool
mcp_tool = MCPStreamableHTTPTool(
name="Product Recommendation MCP",
url=MCP_SERVER_URL,
)
# Initialize Azure OpenAI Chat Client
chat_client = AzureOpenAIChatClient(
endpoint=os.getenv("AZURE_OPENAI_ENDPOINT"),
deployment_name=os.getenv("AZURE_OPENAI_DEPLOYMENT_NAME"),
api_key=os.getenv("AZURE_OPENAI_API_KEY"),
api_version=os.getenv("AZURE_OPENAI_API_VERSION"),
)
# Create the agent with MCP tools
agent = ChatAgent(
chat_client=chat_client,
name="ProductRecommendationAgent",
instructions="""You are a helpful product recommendation assistant for an outdoor equipment store.
You have access to a product catalog through MCP tools that allow you to:
- Search for products using natural language
- Filter by category, brand, and price
- Get product details and descriptions
- List all available categories and brands
When answering questions:
1. Use the appropriate tools to find relevant products
2. Provide clear, detailed product information
3. Include prices, brands, and key features
4. Make helpful recommendations based on customer needs
5. Be concise but informative
Always cite specific products with their names, prices, and relevant details.""",
)
print("✅ Microsoft Agent Framework initialized successfully!")
print(f"🤖 Agent: {agent.name}")
print(f"🔧 MCP Tool: {mcp_tool.name}")
async def test_all_mcp_methods():
"""Test all MCP server methods using Microsoft Agent Framework"""
# Initialize the agent
await initialize_agent()
print_section("PRODUCT RECOMMENDATION MCP CLIENT - TESTING WITH MAF")
# Test cases - same as the original client
test_cases = [
{
"title": "TEST 1: Search Products (Natural Language)",
"query": "I'm looking for a waterproof tent for camping"
},
{
"title": "TEST 2: Get All Categories",
"query": "What product categories are available in the catalog?"
},
{
"title": "TEST 3: Get All Brands",
"query": "What brands do you have available?"
},
{
"title": "TEST 4: Search Products by Category",
"query": "Show me hiking boots from the Hiking Footwear category"
},
{
"title": "TEST 5: Search Products by Brand",
"query": "What products does HikeMate offer?"
},
{
"title": "TEST 6: Search Products by Category and Brand",
"query": "Find tents from OutdoorLiving brand"
},
{
"title": "TEST 7: Get Product Description",
"query": "Tell me everything about the TrailMaster X4 Tent"
},
{
"title": "TEST 8: Search with Price Filter (Less Than)",
"query": "Find hiking clothing items that cost less than 100 USD"
},
{
"title": "TEST 9: Search with Price Filter (Greater Than)",
"query": "Show me premium products over 200 USD"
},
{
"title": "TEST 10: Complex Query",
"query": "I need an affordable backpack for hiking, preferably under 100 USD"
},
]
# Run all test cases
for test_case in test_cases:
print_section(test_case["title"])
query = test_case["query"]
try:
# Run the agent with the query and MCP tools
result = await agent.run(query, tools=mcp_tool)
# Extract response text
response_text = result.text if hasattr(result, 'text') else str(result)
print_result(query, response_text)
except Exception as e:
print(f"\n❌ Error processing query: {str(e)}")
import traceback
traceback.print_exc()
print_section("ALL TESTS COMPLETED SUCCESSFULLY! ✨")
print("\n🎉 Microsoft Agent Framework with MCP is working perfectly!")
print("📊 All 8 MCP methods have been tested with various scenarios.")
print("\n" + "="*80 + "\n")
def main():
"""Main entry point"""
print("\n" + "🚀 "*20)
print(" PRODUCT RECOMMENDATION - MICROSOFT AGENT FRAMEWORK TEST")
print("🚀 "*20 + "\n")
try:
asyncio.run(test_all_mcp_methods())
except KeyboardInterrupt:
print("\n\n⚠️ Test interrupted by user")
except Exception as e:
print(f"\n\n❌ Error during testing: {e}")
import traceback
traceback.print_exc()
if __name__ == "__main__":
main()