MCP Integration & External Tool Connectivity in Microsoft Agent Framework – Part 9

import asyncio
import os
from dataclasses import dataclass
from typing import List, Optional, Any
import aiohttp
from agent_framework.mcp import MCPClient, MCPServerConfig
from agent_framework.azure import AzureOpenAIResponsesClient
from azure.identity import AzureCliCredential

@dataclass
class MCPTool:
    name: str
    description: str
    parameters: dict

async def connect_to_mcp_server():
    """Connect to an MCP server and discover available tools."""
    
    # Configure MCP server connection
    mcp_config = MCPServerConfig(
        url="http://localhost:8080",
        auth_token=os.getenv("MCP_TOKEN"),
        timeout=30,
        retry_attempts=3
    )
    
    # Create MCP client
    mcp_client = MCPClient(config=mcp_config)
    
    # Connect and discover capabilities
    await mcp_client.connect()
    
    # List available tools
    tools: List[MCPTool] = await mcp_client.list_tools()
    print(f"Discovered {len(tools)} tools from MCP server:")
    for tool in tools:
        print(f"  - {tool.name}: {tool.description}")
    
    # List available resources (if any)
    resources = await mcp_client.list_resources()
    print(f"\nDiscovered {len(resources)} resources:")
    for resource in resources:
        print(f"  - {resource.uri}: {resource.description}")
    
    # List available prompts (if any)
    prompts = await mcp_client.list_prompts()
    print(f"\nDiscovered {len(prompts)} prompt templates:")
    for prompt in prompts:
        print(f"  - {prompt.name}: {prompt.description}")
    
    return mcp_client

async def create_mcp_enabled_agent():
    """Create an agent with MCP tools automatically registered."""
    
    # Connect to MCP server
    mcp_client = await connect_to_mcp_server()
    
    # Create agent with MCP tools
    agent = AzureOpenAIResponsesClient(
        credential=AzureCliCredential()
    ).create_agent(
        name="MCPEnabledAgent",
        instructions="""
            You are a helpful assistant with access to external tools via MCP.
            Use the available tools to help users with their requests.
            Always explain what tool you're using and why.
        """,
        mcp_clients=[mcp_client]  # Tools are auto-discovered and registered
    )
    
    return agent

async def main():
    agent = await create_mcp_enabled_agent()
    
    # The agent can now use any tool from the MCP server
    result = await agent.run(
        "Search for recent issues in the microsoft/agent-framework GitHub repo"
    )
    print(f"\nAgent response:\n{result.text}")

if __name__ == "__main__":
    asyncio.run(main())

import asyncio
import os
from agent_framework.mcp import MCPClient, MCPServerConfig
from agent_framework.azure import AzureOpenAIResponsesClient
from azure.identity import AzureCliCredential

async def create_multi_mcp_agent():
    """
    Create an agent connected to multiple MCP servers.
    Each server provides different capabilities.
    """
    
    # GitHub MCP - for code and issue management
    github_mcp = MCPClient(MCPServerConfig(
        url="http://localhost:8081",
        auth_token=os.getenv("GITHUB_MCP_TOKEN"),
        timeout=30
    ))
    await github_mcp.connect()
    print(f"GitHub MCP: {len(await github_mcp.list_tools())} tools")
    
    # Slack MCP - for team communication
    slack_mcp = MCPClient(MCPServerConfig(
        url="http://localhost:8082",
        auth_token=os.getenv("SLACK_MCP_TOKEN"),
        timeout=30
    ))
    await slack_mcp.connect()
    print(f"Slack MCP: {len(await slack_mcp.list_tools())} tools")
    
    # PostgreSQL MCP - for database queries
    db_mcp = MCPClient(MCPServerConfig(
        url="http://localhost:8083",
        auth_token=os.getenv("DB_MCP_TOKEN"),
        timeout=60  # Longer timeout for DB queries
    ))
    await db_mcp.connect()
    print(f"Database MCP: {len(await db_mcp.list_tools())} tools")
    
    # Filesystem MCP - for document access
    fs_mcp = MCPClient(MCPServerConfig(
        url="http://localhost:8084",
        timeout=30
    ))
    await fs_mcp.connect()
    print(f"Filesystem MCP: {len(await fs_mcp.list_tools())} tools")
    
    # Create agent with all MCP connections
    agent = AzureOpenAIResponsesClient(
        credential=AzureCliCredential()
    ).create_agent(
        name="EnterpriseAgent",
        instructions="""
            You are an enterprise assistant with access to multiple systems:
            
            - GitHub: Search code, list issues, create PRs
            - Slack: Send messages, search channels, manage notifications
            - Database: Query customer data, generate reports
            - Filesystem: Search and read documents
            
            Choose the appropriate tool based on the user's request.
            You can combine multiple tools to complete complex tasks.
        """,
        mcp_clients=[github_mcp, slack_mcp, db_mcp, fs_mcp]
    )
    
    return agent

async def demo_multi_mcp():
    agent = await create_multi_mcp_agent()
    
    # Complex task using multiple MCP servers
    result = await agent.run("""
        1. Find the latest critical bug reports in GitHub for our main repo
        2. Query the database to see how many customers are affected
        3. Send a summary to the #engineering Slack channel
    """)
    
    print(f"\nResult:\n{result.text}")

if __name__ == "__main__":
    asyncio.run(demo_multi_mcp())

import asyncio
from dataclasses import dataclass
from typing import Any, List, Optional
import json

from mcp.server import Server
from mcp.types import Tool, TextContent, Resource

# Initialize MCP server
server = Server("company-internal-tools")

# ============================================================
# TOOL DEFINITIONS
# ============================================================

@server.tool("lookup_employee")
async def lookup_employee(employee_id: str) -> str:
    """
    Look up employee information by ID.
    
    Args:
        employee_id: The employee's unique identifier (e.g., EMP-12345)
    
    Returns:
        Employee information including name, department, and contact
    """
    # Connect to internal HR system (mock implementation)
    employees = {
        "EMP-001": {"name": "Alice Johnson", "dept": "Engineering", "email": "alice@company.com"},
        "EMP-002": {"name": "Bob Smith", "dept": "Product", "email": "bob@company.com"},
        "EMP-003": {"name": "Carol Williams", "dept": "Sales", "email": "carol@company.com"},
    }
    
    employee = employees.get(employee_id)
    if employee:
        return json.dumps({
            "id": employee_id,
            "name": employee["name"],
            "department": employee["dept"],
            "email": employee["email"],
            "status": "active"
        }, indent=2)
    
    return f"Employee {employee_id} not found"

@server.tool("submit_ticket")
async def submit_ticket(
    title: str,
    description: str,
    priority: str = "medium",
    assignee: Optional[str] = None
) -> str:
    """
    Submit an internal support ticket.
    
    Args:
        title: Brief title for the ticket
        description: Detailed description of the issue
        priority: Priority level (low, medium, high, critical)
        assignee: Optional employee ID to assign the ticket
    
    Returns:
        Ticket confirmation with ticket ID
    """
    import uuid
    from datetime import datetime
    
    ticket_id = f"TKT-{uuid.uuid4().hex[:8].upper()}"
    
    # In production, save to ticketing system
    ticket = {
        "id": ticket_id,
        "title": title,
        "description": description,
        "priority": priority,
        "assignee": assignee,
        "status": "open",
        "created": datetime.now().isoformat()
    }
    
    return json.dumps({
        "success": True,
        "ticket": ticket,
        "message": f"Ticket {ticket_id} created successfully"
    }, indent=2)

@server.tool("query_metrics")
async def query_metrics(
    metric_name: str,
    time_range: str = "24h"
) -> str:
    """
    Query internal metrics and KPIs.
    
    Args:
        metric_name: Name of the metric (e.g., sales, uptime, errors)
        time_range: Time range (1h, 24h, 7d, 30d)
    
    Returns:
        Metric data with values and trends
    """
    # Mock metrics data
    metrics = {
        "sales": {"value": 125000, "unit": "USD", "change": "+12%"},
        "uptime": {"value": 99.98, "unit": "%", "change": "+0.02%"},
        "errors": {"value": 23, "unit": "count", "change": "-15%"},
        "response_time": {"value": 145, "unit": "ms", "change": "-8%"},
    }
    
    metric = metrics.get(metric_name.lower())
    if metric:
        return json.dumps({
            "metric": metric_name,
            "time_range": time_range,
            "current_value": f"{metric['value']} {metric['unit']}",
            "trend": metric["change"],
            "status": "healthy"
        }, indent=2)
    
    return f"Metric '{metric_name}' not found. Available: {list(metrics.keys())}"

@server.tool("search_documents")
async def search_documents(
    query: str,
    doc_type: Optional[str] = None,
    limit: int = 10
) -> str:
    """
    Search internal document repository.
    
    Args:
        query: Search query
        doc_type: Optional filter by document type (policy, procedure, template)
        limit: Maximum number of results
    
    Returns:
        List of matching documents with links
    """
    # Mock document search
    results = [
        {"title": f"Policy: {query} Guidelines", "type": "policy", "url": "/docs/policy-001"},
        {"title": f"Procedure: How to {query}", "type": "procedure", "url": "/docs/proc-001"},
        {"title": f"Template: {query} Request Form", "type": "template", "url": "/docs/tmpl-001"},
    ]
    
    if doc_type:
        results = [r for r in results if r["type"] == doc_type]
    
    return json.dumps({
        "query": query,
        "results": results[:limit],
        "total_count": len(results)
    }, indent=2)

# ============================================================
# RESOURCE DEFINITIONS
# ============================================================

@server.resource("company://org-chart")
async def get_org_chart() -> str:
    """Provide the company org chart as a resource."""
    return json.dumps({
        "ceo": "Jane Doe",
        "departments": [
            {"name": "Engineering", "head": "Alice Johnson", "headcount": 45},
            {"name": "Product", "head": "Bob Smith", "headcount": 12},
            {"name": "Sales", "head": "Carol Williams", "headcount": 28},
        ]
    }, indent=2)

@server.resource("company://holidays-2025")
async def get_holidays() -> str:
    """Provide company holidays for 2025."""
    return json.dumps({
        "year": 2025,
        "holidays": [
            {"date": "2025-01-01", "name": "New Year's Day"},
            {"date": "2025-07-04", "name": "Independence Day"},
            {"date": "2025-12-25", "name": "Christmas Day"},
        ]
    }, indent=2)

# ============================================================
# SERVER STARTUP
# ============================================================

if __name__ == "__main__":
    import sys
    
    print("Starting Company Internal Tools MCP Server...")
    print(f"Tools: lookup_employee, submit_ticket, query_metrics, search_documents")
    print(f"Resources: company://org-chart, company://holidays-2025")
    
    # Run on stdio (for local) or TCP (for remote)
    if "--tcp" in sys.argv:
        server.run(transport="tcp", port=8080)
    else:
        server.run(transport="stdio")

import asyncio
import os
from msal import ConfidentialClientApplication
from agent_framework.mcp import MCPClient, MCPServerConfig
from agent_framework.azure import AzureOpenAIResponsesClient
from azure.identity import DefaultAzureCredential

class Microsoft365Integration:
    """
    Enterprise integration with Microsoft 365 services via MCP.
    Provides access to SharePoint, Outlook, Teams, and OneDrive.
    """
    
    def __init__(self):
        self.tenant_id = os.getenv("AZURE_TENANT_ID")
        self.client_id = os.getenv("AZURE_CLIENT_ID")
        self.client_secret = os.getenv("AZURE_CLIENT_SECRET")
        
        # MSAL client for M365 authentication
        self.msal_app = ConfidentialClientApplication(
            client_id=self.client_id,
            client_credential=self.client_secret,
            authority=f"https://login.microsoftonline.com/{self.tenant_id}"
        )
    
    async def get_access_token(self, scopes: list) -> str:
        """Get access token for Microsoft Graph API."""
        result = self.msal_app.acquire_token_for_client(scopes=scopes)
        if "access_token" in result:
            return result["access_token"]
        raise Exception(f"Failed to get token: {result.get('error_description')}")
    
    async def create_sharepoint_mcp(self, site_url: str) -> MCPClient:
        """Create MCP client for SharePoint operations."""
        token = await self.get_access_token(["https://graph.microsoft.com/.default"])
        
        return MCPClient(MCPServerConfig(
            url="http://localhost:8090",  # SharePoint MCP server
            auth_token=token,
            headers={"X-SharePoint-Site": site_url},
            timeout=60
        ))
    
    async def create_outlook_mcp(self, user_email: str) -> MCPClient:
        """Create MCP client for Outlook operations."""
        token = await self.get_access_token(["https://graph.microsoft.com/.default"])
        
        return MCPClient(MCPServerConfig(
            url="http://localhost:8091",  # Outlook MCP server
            auth_token=token,
            headers={"X-User-Email": user_email},
            timeout=30
        ))
    
    async def create_teams_mcp(self) -> MCPClient:
        """Create MCP client for Teams operations."""
        token = await self.get_access_token(["https://graph.microsoft.com/.default"])
        
        return MCPClient(MCPServerConfig(
            url="http://localhost:8092",  # Teams MCP server
            auth_token=token,
            timeout=30
        ))

async def create_m365_agent():
    """Create an agent with full Microsoft 365 integration."""
    
    m365 = Microsoft365Integration()
    
    # Connect to M365 MCP servers
    sharepoint = await m365.create_sharepoint_mcp(
        site_url="https://company.sharepoint.com/sites/docs"
    )
    await sharepoint.connect()
    print(f"SharePoint connected: {len(await sharepoint.list_tools())} tools")
    
    outlook = await m365.create_outlook_mcp(
        user_email="agent@company.com"
    )
    await outlook.connect()
    print(f"Outlook connected: {len(await outlook.list_tools())} tools")
    
    teams = await m365.create_teams_mcp()
    await teams.connect()
    print(f"Teams connected: {len(await teams.list_tools())} tools")
    
    # Create agent with M365 capabilities
    agent = AzureOpenAIResponsesClient(
        credential=DefaultAzureCredential()
    ).create_agent(
        name="M365EnterpriseAgent",
        instructions="""
            You are an enterprise assistant with Microsoft 365 integration.
            
            Available capabilities:
            
            SharePoint:
            - Search documents and files
            - Read file contents
            - List recent documents
            - Get document metadata
            
            Outlook:
            - Search emails
            - Read email content
            - Send emails (with user confirmation)
            - Manage calendar
            
            Teams:
            - Send channel messages
            - Search conversations
            - List team members
            - Schedule meetings
            
            Always confirm before sending emails or messages.
            Respect user privacy and data sensitivity.
        """,
        mcp_clients=[sharepoint, outlook, teams]
    )
    
    return agent

async def demo_m365_agent():
    agent = await create_m365_agent()
    
    # Example: Cross-platform task
    result = await agent.run("""
        1. Find the latest quarterly report in SharePoint
        2. Summarize the key findings
        3. Draft an email to the leadership team with the summary
        (Don't send yet - just show me the draft)
    """)
    
    print(f"\nResult:\n{result.text}")

if __name__ == "__main__":
    asyncio.run(demo_m365_agent())