Skip to main content

Azure AI Foundry Agent Service

Connect Unizo MCP to Azure AI Foundry Agent Service with native MCP integration.

Overview​

Azure AI Foundry Agent Service provides native MCP support, allowing agents to connect to Unizo's MCP server for enterprise-grade AI workflows with comprehensive security and business platform integration.

Official Documentation

Prerequisites​

  • Azure AI Foundry workspace
  • Azure OpenAI deployment
  • Unizo API credentials

Quick Start​

Connect to Unizo MCP via Azure AI Foundry:

from azure.ai.projects import AIProjectClient
from azure.identity import DefaultAzureCredential
import os

# Create Azure AI project client
project_client = AIProjectClient.from_connection_string(
    credential=DefaultAzureCredential(),
    conn_str="<your-connection-string>"
)

# Register Unizo MCP server
mcp_connection = project_client.connections.create_or_update(
    connection_name="unizo-mcp",
    properties={
        "connectionType": "MCP",
        "serverUrl": "https://api.unizo.ai/mcp",
        "headers": {
            "Authorization": f"Bearer {os.getenv('UNIZO_API_KEY')}",
            "x-mcp-scopes": "edr,vms,ticketing",
            "Content-Type": "application/json",
            "Accept": "application/json,text/event-stream",
            "MCP-Protocol-Version": "2025-06-18"
        }
    }
)

# Create agent with Unizo MCP tools
agent = project_client.agents.create_agent(
    model="gpt-4o",
    name="Security Operations Assistant",
    instructions="You help analyze security data and manage incidents using connected security platforms",
    tools=[{
        "type": "mcp",
        "mcp_server": "unizo-mcp"
    }]
)

# Create thread and run
thread = project_client.agents.create_thread()
message = project_client.agents.create_message(
    thread_id=thread.id,
    role="user",
    content="Check for critical security incidents in SentinelOne"
)

run = project_client.agents.create_and_process_run(
    thread_id=thread.id,
    assistant_id=agent.id
)

print(run.status)

Portal Configuration​

You can also configure MCP connections through the Azure AI Foundry portal:

  1. Navigate to your AI Foundry workspace
  2. Go to Connections → Add Connection
  3. Select Model Context Protocol (MCP)
  4. Enter Unizo MCP details:
    • Server URL: https://api.unizo.ai/mcp
    • Headers: Add authentication headers
    • Server Label: unizo-mcp

Learn more in Azure docs

Advanced Usage​

Security Operations Agent​

Create a specialized security agent with comprehensive monitoring capabilities:

from azure.ai.projects import AIProjectClient
from azure.identity import DefaultAzureCredential
import os

def create_security_agent(project_client: AIProjectClient):
    """Create a security operations agent with EDR, VMS, and ticketing capabilities"""
    
    # Register Unizo MCP with security-focused scopes
    mcp_connection = project_client.connections.create_or_update(
        connection_name="unizo-security-mcp",
        properties={
            "connectionType": "MCP",
            "serverUrl": "https://api.unizo.ai/mcp",
            "headers": {
                "Authorization": f"Bearer {os.getenv('UNIZO_API_KEY')}",
                "x-mcp-scopes": "edr,vms,ticketing",
                "Content-Type": "application/json",
                "Accept": "application/json,text/event-stream",
                "MCP-Protocol-Version": "2025-06-18"
            }
        }
    )

    # Create security-focused agent
    agent = project_client.agents.create_agent(
        model="gpt-4o",
        name="Security Operations Manager",
        instructions="""You are a security operations assistant with access to:
        - Endpoint Detection & Response (EDR) systems
        - Vulnerability Management systems
        - Incident ticketing systems
        
        Your role is to:
        1. Monitor security incidents and threats
        2. Analyze vulnerability data and prioritize risks
        3. Create and manage security tickets
        4. Provide actionable security recommendations
        5. Coordinate incident response activities
        
        Always prioritize critical and high-severity security events.""",
        tools=[{
            "type": "mcp",
            "mcp_server": "unizo-security-mcp"
        }]
    )
    
    return agent

# Usage
project_client = AIProjectClient.from_connection_string(
    credential=DefaultAzureCredential(),
    conn_str=os.getenv("AZURE_AI_PROJECT_CONNECTION_STRING")
)

security_agent = create_security_agent(project_client)

# Create security monitoring thread
thread = project_client.agents.create_thread()

# Run security analysis
security_queries = [
    "Check for critical malware detections in the last 24 hours",
    "List high and critical vulnerabilities that need immediate attention",
    "Create tickets for any critical security findings"
]

for query in security_queries:
    message = project_client.agents.create_message(
        thread_id=thread.id,
        role="user",
        content=query
    )
    
    run = project_client.agents.create_and_process_run(
        thread_id=thread.id,
        assistant_id=security_agent.id
    )
    
    # Wait for completion and get response
    while run.status in ["queued", "in_progress", "requires_action"]:
        time.sleep(1)
        run = project_client.agents.get_run(thread.id, run.id)
    
    if run.status == "completed":
        messages = project_client.agents.list_messages(thread.id)
        print(f"Security Analysis: {messages.data[0].content[0].text.value}")

Business Intelligence Agent​

Create an agent for comprehensive business operations insights:

def create_business_intelligence_agent(project_client: AIProjectClient):
    """Create a business intelligence agent with multi-service access"""
    
    # Register Unizo MCP with comprehensive business scopes
    mcp_connection = project_client.connections.create_or_update(
        connection_name="unizo-business-mcp",
        properties={
            "connectionType": "MCP",
            "serverUrl": "https://api.unizo.ai/mcp",
            "headers": {
                "Authorization": f"Bearer {os.getenv('UNIZO_API_KEY')}",
                "x-mcp-scopes": "edr,vms,ticketing,scm,communications,observability,infra",
                "Content-Type": "application/json",
                "Accept": "application/json,text/event-stream",
                "MCP-Protocol-Version": "2025-06-18"
            }
        }
    )

    # Create comprehensive business agent
    agent = project_client.agents.create_agent(
        model="gpt-4o",
        name="Business Intelligence Analyst",
        instructions="""You are a business intelligence analyst with access to:
        - Security systems (EDR, vulnerability management)
        - IT service management (ticketing)
        - Development tools (source code management)
        - Team communications
        - Infrastructure monitoring
        - Observability tools
        
        Provide insights on:
        - Security posture and risk assessment
        - Operational efficiency and performance metrics
        - Development team productivity
        - Infrastructure health and capacity
        - Cross-functional collaboration patterns
        
        Generate actionable recommendations for business improvement.""",
        tools=[{
            "type": "mcp",
            "mcp_server": "unizo-business-mcp"
        }]
    )
    
    return agent

# Create BI agent and analyze business metrics
bi_agent = create_business_intelligence_agent(project_client)

thread = project_client.agents.create_thread()

bi_queries = [
    "Analyze our security posture and provide a risk assessment",
    "Review development team productivity metrics from SCM data",
    "Check infrastructure health and identify capacity issues",
    "Generate a weekly business operations summary"
]

for query in bi_queries:
    message = project_client.agents.create_message(
        thread_id=thread.id,
        role="user", 
        content=query
    )
    
    run = project_client.agents.create_and_process_run(
        thread_id=thread.id,
        assistant_id=bi_agent.id
    )
    
    # Process results...

Multi-Agent Workflow​

Coordinate multiple specialized agents for complex workflows:

import asyncio
from typing import Dict, List

class MultiAgentWorkflow:
    def __init__(self, project_client: AIProjectClient):
        self.project_client = project_client
        self.agents = {}
        
    def create_specialized_agents(self):
        """Create multiple specialized agents"""
        
        # Security Operations Agent
        self.agents['security'] = self.project_client.agents.create_agent(
            model="gpt-4o",
            name="Security Specialist",
            instructions="Focus on security incidents, threats, and vulnerabilities",
            tools=[{
                "type": "mcp",
                "mcp_server": self._create_mcp_connection("security", "edr,vms,ticketing")
            }]
        )
        
        # Infrastructure Agent  
        self.agents['infrastructure'] = self.project_client.agents.create_agent(
            model="gpt-4o",
            name="Infrastructure Engineer", 
            instructions="Monitor infrastructure health and performance",
            tools=[{
                "type": "mcp",
                "mcp_server": self._create_mcp_connection("infra", "infra,observability")
            }]
        )
        
        # Development Operations Agent
        self.agents['devops'] = self.project_client.agents.create_agent(
            model="gpt-4o",
            name="DevOps Engineer",
            instructions="Analyze development processes and code management",
            tools=[{
                "type": "mcp", 
                "mcp_server": self._create_mcp_connection("devops", "scm,ticketing")
            }]
        )
        
    def _create_mcp_connection(self, name: str, scopes: str) -> str:
        """Create MCP connection with specific scopes"""
        connection_name = f"unizo-{name}-mcp"
        
        self.project_client.connections.create_or_update(
            connection_name=connection_name,
            properties={
                "connectionType": "MCP",
                "serverUrl": "https://api.unizo.ai/mcp",
                "headers": {
                    "Authorization": f"Bearer {os.getenv('UNIZO_API_KEY')}",
                    "x-mcp-scopes": scopes,
                    "Content-Type": "application/json",
                    "Accept": "application/json,text/event-stream",
                    "MCP-Protocol-Version": "2025-06-18"
                }
            }
        )
        
        return connection_name
        
    def run_coordinated_analysis(self) -> Dict[str, str]:
        """Run coordinated analysis across all agents"""
        
        tasks = {
            'security': "Analyze current security threats and create priority action plan",
            'infrastructure': "Assess infrastructure health and resource utilization",
            'devops': "Review development velocity and identify bottlenecks"
        }
        
        results = {}
        
        for agent_type, query in tasks.items():
            thread = self.project_client.agents.create_thread()
            
            message = self.project_client.agents.create_message(
                thread_id=thread.id,
                role="user",
                content=query
            )
            
            run = self.project_client.agents.create_and_process_run(
                thread_id=thread.id,
                assistant_id=self.agents[agent_type].id
            )
            
            # Wait for completion and collect results
            while run.status in ["queued", "in_progress"]:
                time.sleep(1)
                run = self.project_client.agents.get_run(thread.id, run.id)
                
            if run.status == "completed":
                messages = self.project_client.agents.list_messages(thread.id)
                results[agent_type] = messages.data[0].content[0].text.value
                
        return results

# Usage
workflow = MultiAgentWorkflow(project_client)
workflow.create_specialized_agents()
analysis_results = workflow.run_coordinated_analysis()

# Generate consolidated report
for agent_type, result in analysis_results.items():
    print(f"\n{agent_type.upper()} ANALYSIS:")
    print(result)

Streaming Agent Responses​

Handle real-time agent responses for interactive applications:

def create_streaming_agent(project_client: AIProjectClient):
    """Create agent with streaming response handling"""
    
    # Create agent with streaming support
    agent = project_client.agents.create_agent(
        model="gpt-4o",
        name="Interactive Security Assistant",
        instructions="Provide real-time security analysis and recommendations",
        tools=[{
            "type": "mcp",
            "mcp_server": "unizo-mcp"
        }]
    )
    
    return agent

def stream_security_analysis(project_client: AIProjectClient, agent_id: str, query: str):
    """Stream security analysis results"""
    
    thread = project_client.agents.create_thread()
    
    message = project_client.agents.create_message(
        thread_id=thread.id,
        role="user",
        content=query
    )
    
    # Create run with streaming
    run = project_client.agents.create_run(
        thread_id=thread.id,
        assistant_id=agent_id
    )
    
    # Stream responses
    while run.status not in ["completed", "failed", "cancelled"]:
        time.sleep(0.5)
        run = project_client.agents.get_run(thread.id, run.id)
        
        if run.status == "in_progress":
            # Get partial messages if available
            messages = project_client.agents.list_messages(thread.id)
            if messages.data:
                latest_message = messages.data[0]
                if latest_message.role == "assistant":
                    # Process streaming content
                    content = latest_message.content[0].text.value
                    print(f"Streaming: {content}")
                    
    # Get final result
    if run.status == "completed":
        messages = project_client.agents.list_messages(thread.id)
        final_response = messages.data[0].content[0].text.value
        return final_response
        
    return None

# Usage
streaming_agent = create_streaming_agent(project_client)
result = stream_security_analysis(
    project_client, 
    streaming_agent.id,
    "Provide real-time analysis of current security threats and recommended actions"
)

Available Scopes​

Configure x-mcp-scopes to control which Unizo services your agents can access:

  • edr: Endpoint Detection & Response tools
  • vms: Vulnerability Management tools
  • ticketing: Incident and ticket management tools
  • scm: Source Code Management tools
  • communications: Team communication tools
  • observability: Monitoring and logging tools
  • infra: Infrastructure management tools

Error Handling​

Implement robust error handling for production Azure deployments:

from azure.ai.projects.exceptions import AIProjectException
import logging

def create_agent_with_error_handling(project_client: AIProjectClient):
    """Create agent with comprehensive error handling"""
    
    try:
        # Attempt to create MCP connection
        mcp_connection = project_client.connections.create_or_update(
            connection_name="unizo-mcp",
            properties={
                "connectionType": "MCP",
                "serverUrl": "https://api.unizo.ai/mcp",
                "headers": {
                    "Authorization": f"Bearer {os.getenv('UNIZO_API_KEY')}",
                    "x-mcp-scopes": "edr,vms,ticketing",
                    "Content-Type": "application/json",
                    "Accept": "application/json,text/event-stream",
                    "MCP-Protocol-Version": "2025-06-18"
                }
            }
        )
        
        # Create agent
        agent = project_client.agents.create_agent(
            model="gpt-4o",
            name="Security Assistant",
            instructions="Analyze security data with error recovery",
            tools=[{
                "type": "mcp",
                "mcp_server": "unizo-mcp"
            }]
        )
        
        return agent
        
    except AIProjectException as e:
        logging.error(f"Azure AI Project error: {e}")
        raise
        
    except Exception as e:
        logging.error(f"Unexpected error creating agent: {e}")
        raise

def run_agent_with_retry(project_client: AIProjectClient, agent_id: str, query: str, max_retries: int = 3):
    """Run agent with retry logic"""
    
    for attempt in range(max_retries):
        try:
            thread = project_client.agents.create_thread()
            
            message = project_client.agents.create_message(
                thread_id=thread.id,
                role="user",
                content=query
            )
            
            run = project_client.agents.create_and_process_run(
                thread_id=thread.id,
                assistant_id=agent_id
            )
            
            # Wait for completion with timeout
            timeout = 60  # 60 seconds
            elapsed = 0
            
            while run.status in ["queued", "in_progress"] and elapsed < timeout:
                time.sleep(2)
                elapsed += 2
                run = project_client.agents.get_run(thread.id, run.id)
                
            if run.status == "completed":
                messages = project_client.agents.list_messages(thread.id)
                return messages.data[0].content[0].text.value
                
            elif run.status == "failed":
                logging.warning(f"Agent run failed on attempt {attempt + 1}")
                if attempt == max_retries - 1:
                    raise Exception("Agent run failed after all retry attempts")
                    
            else:
                logging.warning(f"Agent run timeout on attempt {attempt + 1}")
                if attempt == max_retries - 1:
                    raise Exception("Agent run timeout after all retry attempts")
                    
        except Exception as e:
            logging.error(f"Error on attempt {attempt + 1}: {e}")
            if attempt == max_retries - 1:
                raise
                
        # Wait before retry
        time.sleep(2 ** attempt)  # Exponential backoff
        
    return None

Environment Variables​

UNIZO_API_KEY=<unizo_api_key>
AZURE_OPENAI_ENDPOINT=your_azure_endpoint
AZURE_OPENAI_KEY=your_azure_key

Resources​