"""
Microsoft Agent Framework Integration
Provides a wrapper around Microsoft Agent Framework for MCP-MAF.
Manages Azure OpenAI connection, tool registration, and agent lifecycle.
"""
import asyncio
import os
from typing import Dict, List, Any, Optional
from .utils.logger import setup_logger
from .tools import AVAILABLE_TOOLS, call_tool as execute_tool
logger = setup_logger(__name__)
class MCPMAFAgent:
"""
Microsoft Agent Framework Wrapper
Manages:
- Azure OpenAI connection
- MCP tool registration
- Agent instance lifecycle
- Conversation handling
"""
def __init__(
self,
azure_endpoint: Optional[str] = None,
api_key: Optional[str] = None,
deployment_name: Optional[str] = None
):
"""
Initialize MCP-MAF Agent
Args:
azure_endpoint: Azure OpenAI endpoint URL
api_key: Azure OpenAI API key
deployment_name: Model deployment name (default: gpt-4.1)
"""
# Configuration
self.endpoint = azure_endpoint or os.getenv("AZURE_OPENAI_ENDPOINT")
self.api_key = api_key or os.getenv("AZURE_OPENAI_API_KEY")
self.deployment = deployment_name or os.getenv("AZURE_AI_MODEL_DEPLOYMENT_NAME", "gpt-4.1")
if not self.endpoint or not self.api_key:
raise ValueError("Azure OpenAI credentials not found. Please set AZURE_OPENAI_ENDPOINT and AZURE_OPENAI_API_KEY.")
# State
self._agent = None
self._client = None
self._tools = {}
self._lock = asyncio.Lock()
self._initialized = False
logger.info("šÆ MCPMAFAgent initialized")
logger.info(f" Endpoint: {self.endpoint}")
logger.info(f" Deployment: {self.deployment}")
async def initialize(self):
"""
Initialize Agent and tools
This method:
1. Initializes MCP tools
2. Creates Azure OpenAI client
3. Registers tools with agent
4. Creates agent instance
"""
async with self._lock:
if self._initialized:
logger.info("ā
Agent already initialized")
return
logger.info("š Initializing MCP-MAF Agent...")
# Step 1: Initialize tools
logger.info("š¦ Step 1: Initializing tools...")
await self._init_tools()
logger.info(f" ā
{len(self._tools)} tools initialized: {list(self._tools.keys())}")
# Step 2: Create Azure OpenAI client
logger.info("š§ Step 2: Creating Azure OpenAI client...")
try:
# Try to import Microsoft Agent Framework
try:
from agent_framework.azure import AzureOpenAIResponsesClient
from azure.core.credentials import AzureKeyCredential
self._client = AzureOpenAIResponsesClient(
credential=AzureKeyCredential(self.api_key),
endpoint=self.endpoint,
deployment_name=self.deployment
)
logger.info(" ā
Using Microsoft Agent Framework")
except ImportError:
# Fallback to OpenAI SDK
logger.warning(" ā ļø Microsoft Agent Framework not available, using OpenAI SDK fallback")
from openai import AsyncAzureOpenAI
self._client = AsyncAzureOpenAI(
api_key=self.api_key,
azure_endpoint=self.endpoint,
api_version="2024-02-01"
)
logger.info(" ā
Using OpenAI SDK fallback")
except Exception as e:
logger.error(f" ā Failed to create client: {e}")
raise
# Step 3: Create agent
logger.info("š¤ Step 3: Creating agent...")
try:
self._agent = await self._create_agent()
logger.info(f" ā
Agent created: {self._agent}")
except Exception as e:
logger.error(f" ā Failed to create agent: {e}")
raise
self._initialized = True
logger.info("š Agent initialization complete!")
async def _init_tools(self):
"""Initialize MCP tools"""
for tool in AVAILABLE_TOOLS:
self._tools[tool.name] = tool
logger.debug(f" š Registered tool: {tool.name}")
async def _create_agent(self):
"""Create agent instance"""
# If using Microsoft Agent Framework
if hasattr(self._client, 'create_agent'):
tools_list = list(self._tools.values())
agent_name = "mcp-maf-agent"
try:
agent = self._client.create_agent(
name=agent_name,
instructions=self._get_instructions(),
tools=tools_list # Register tools
)
logger.info(f" ā
Agent '{agent_name}' created with {len(tools_list)} tools")
return agent
except Exception as e:
logger.info(f" ā¹ļø Agent exists, getting it... ({e})")
return self._client.get_agent(agent_name)
else:
# Using OpenAI SDK fallback
logger.info(" ā¹ļø Using OpenAI SDK fallback mode")
return {"mode": "openai_sdk", "deployment": self.deployment}
def _get_instructions(self) -> str:
"""
Get agent instructions (system prompt)
Returns:
Agent instructions string
"""
return """You are MCP-MAF Agent, a professional AI assistant powered by Microsoft Agent Framework.
## Your Identity
You are a helpful, knowledgeable, and efficient AI assistant designed to help users with:
- Weather information queries
- Mathematical calculations
- Taiwan stock market data
- General information and assistance
## Your Capabilities
You have access to the following tools:
### š¤ļø Weather Tool
- **Function**: `get_weather(location)`
- **Purpose**: Get current weather information for any location
- **Usage**: "What's the weather in Taipei?"
### š§® Calculator Tool
- **Function**: `calculate(expression)`
- **Purpose**: Perform mathematical calculations
- **Usage**: "Calculate 123 * 456"
### š Stock Price Tool
- **Function**: `get_stock_price(stock_code)`
- **Purpose**: Query Taiwan stock prices from TWSE
- **Usage**: "What's TSMC (2330) stock price?"
## Response Guidelines
### 1. Always Use Tools
- Don't guess or make up data
- Always use the appropriate tool to get real-time information
- If a tool fails, explain the error to the user
### 2. Format Responses Nicely
Use clear Markdown formatting:
**Example for Weather**:
```
š¤ļø **Weather in Taipei**
Temperature: 25Ā°C
Condition: Sunny
Humidity: 60%
Have a great day!
```
**Example for Calculations**:
```
š§® **Calculation Result**
Expression: 123 * 456
Result: 56,088
Need another calculation?
```
**Example for Stock Prices**:
```
š **TSMC (2330) Stock Information**
Current Price: NT$ 1,025.00
Change: +15.00 (+1.49%)
Volume: 45,678 shares
Data updated in real-time from Taiwan Stock Exchange.
```
### 3. Be Helpful and Professional
- Provide context with your answers
- Offer to help with related questions
- Be friendly but professional
- If you don't understand, ask for clarification
### 4. Handle Errors Gracefully
- If a tool fails, explain what happened
- Suggest alternatives if possible
- Don't show technical error messages to users
### 5. Multi-step Reasoning
- Break down complex requests into steps
- Use multiple tools if needed
- Explain your reasoning
## Important Notes
- Always prioritize accuracy over speed
- Use tools for real-time data
- Format responses for readability
- Be helpful and professional
Let's help users efficiently!
"""
async def run(self, message: str) -> str:
"""
Execute agent with a message
Args:
message: User message
Returns:
Agent response text
"""
if not self._initialized:
await self.initialize()
logger.info(f"š¬ Processing message: {message[:100]}...")
try:
# If using Microsoft Agent Framework
if hasattr(self._agent, 'run'):
response = await self._agent.run(message)
result = str(response) if response else ""
logger.info(f"ā
Response generated ({len(result)} chars)")
return result
else:
# OpenAI SDK fallback
logger.info(" ā¹ļø Using OpenAI SDK fallback for chat")
response = await self._client.chat.completions.create(
model=self.deployment,
messages=[
{"role": "system", "content": self._get_instructions()},
{"role": "user", "content": message}
]
)
result = response.choices[0].message.content
logger.info(f"ā
Response generated ({len(result)} chars)")
return result
except Exception as e:
error_msg = f"Error processing request: {str(e)}"
logger.error(f"ā {error_msg}")
return error_msg
async def get_tools_list(self) -> List[Dict[str, Any]]:
"""
Get list of available tools
Returns:
List of tool definitions for MCP tools/list
"""
if not self._initialized:
await self.initialize()
return [tool.to_dict() for tool in self._tools.values()]
async def call_tool(self, tool_name: str, arguments: Dict[str, Any]) -> Dict[str, Any]:
"""
Call a specific tool
Args:
tool_name: Tool name
arguments: Tool arguments
Returns:
Tool execution result
"""
if not self._initialized:
await self.initialize()
logger.info(f"šØ Tool call: {tool_name}({arguments})")
if tool_name not in self._tools:
error_msg = f"Tool not found: {tool_name}"
logger.error(f"ā {error_msg}")
raise ValueError(error_msg)
try:
result = await execute_tool(tool_name, arguments)
logger.info(f"ā
Tool executed successfully")
return result
except Exception as e:
error_msg = f"Tool execution failed: {str(e)}"
logger.error(f"ā {error_msg}")
raise
def is_initialized(self) -> bool:
"""Check if agent is initialized"""
return self._initialized
async def close(self):
"""Close agent and cleanup resources"""
logger.info("š Closing agent...")
# Cleanup if needed
self._initialized = False
logger.info("ā
Agent closed")
