Add support for custom MCP server URLs

libraries/microsoft-agents-a365-tooling-extensions-agentframework/microsoft_agents_a365/tooling/extensions/agentframework/services/mcp_tool_registration_service.py

@@ -97,9 +97,8 @@ async def add_tool_servers_to_agent(
             # Add servers as MCPStreamableHTTPTool instances
             for config in server_configs:
                 try:
-                    server_url = getattr(config, "server_url", None) or getattr(
-                        config, "mcp_server_unique_name", None
-                    )
+                    # Use custom URL if provided, otherwise use the unique name
+                    server_url = config.url if config.url else config.mcp_server_unique_name
                     if not server_url:
                         self._logger.warning(f"MCP server config missing server_url: {config}")
                         continue
@@ -115,7 +114,7 @@ async def add_tool_servers_to_agent(
                         self._orchestrator_name
                     )
 
-                    server_name = getattr(config, "mcp_server_name", "Unknown")
+                    server_name = config.mcp_server_name
 
                     # Create and configure MCPStreamableHTTPTool
                     mcp_tools = MCPStreamableHTTPTool(
@@ -134,9 +133,8 @@ async def add_tool_servers_to_agent(
                     self._logger.info(f"Added MCP plugin '{server_name}' to agent tools")
 
                 except Exception as tool_ex:
-                    server_name = getattr(config, "mcp_server_name", "Unknown")
                     self._logger.warning(
-                        f"Failed to create MCP plugin for {server_name}: {tool_ex}"
+                        f"Failed to create MCP plugin for {config.mcp_server_name}: {tool_ex}"
                     )
                     continue
 

libraries/microsoft-agents-a365-tooling-extensions-azureaifoundry/microsoft_agents_a365/tooling/extensions/azureaifoundry/services/mcp_tool_registration_service.py

@@ -178,8 +178,11 @@ async def _get_mcp_tool_definitions_and_resources(
                 else server.mcp_server_name
             )
 
+            # Use custom URL if provided, otherwise use the unique name
+            server_url = server.url if server.url else server.mcp_server_unique_name
+
             # Create MCP tool using Azure Foundry SDK
-            mcp_tool = McpTool(server_label=server_label, server_url=server.mcp_server_unique_name)
+            mcp_tool = McpTool(server_label=server_label, server_url=server_url)
 
             # Configure the tool
             mcp_tool.set_approval_mode("never")

libraries/microsoft-agents-a365-tooling-extensions-openai/microsoft_agents_a365/tooling/extensions/openai/mcp_tool_registration_service.py

@@ -101,9 +101,13 @@ async def add_tool_servers_to_agent(
         # Convert MCP server configs to MCPServerInfo objects
         mcp_servers_info = []
         for server_config in mcp_server_configs:
+            # Use custom URL if provided, otherwise use the unique name
+            server_url = (
+                server_config.url if server_config.url else server_config.mcp_server_unique_name
+            )
             server_info = MCPServerInfo(
                 name=server_config.mcp_server_name,
-                url=server_config.mcp_server_unique_name,
+                url=server_url,
             )
             mcp_servers_info.append(server_info)
 

libraries/microsoft-agents-a365-tooling-extensions-semantickernel/microsoft_agents_a365/tooling/extensions/semantickernel/services/mcp_tool_registration_service.py

@@ -125,9 +125,12 @@ async def add_tool_servers_to_agent(
                     self._orchestrator_name
                 )
 
+                # Use custom URL if provided, otherwise use the unique name  
+                server_url = server.url if server.url else server.mcp_server_unique_name
+
                 plugin = MCPStreamableHttpPlugin(
                     name=server.mcp_server_name,
-                    url=server.mcp_server_unique_name,
+                    url=server_url,
                     headers=headers,
                 )
 

libraries/microsoft-agents-a365-tooling/microsoft_agents_a365/tooling/models/mcp_server_config.py

@@ -5,6 +5,7 @@
 """
 
 from dataclasses import dataclass
+from typing import Optional
 
 
 @dataclass
@@ -19,6 +20,10 @@ class MCPServerConfig:
     #: Gets or sets the unique name of the MCP server.
     mcp_server_unique_name: str
 
+    #: Gets or sets the custom URL for the MCP server. If provided, this URL will be used
+    #: instead of constructing the URL from the base URL and unique name.
+    url: Optional[str] = None
+
     def __post_init__(self):
         """Validate the configuration after initialization."""
         if not self.mcp_server_name:

libraries/microsoft-agents-a365-tooling/microsoft_agents_a365/tooling/services/mcp_tool_server_configuration_service.py

@@ -398,10 +398,15 @@ def _parse_manifest_server_config(
             if not self._validate_server_strings(name, server_name):
                 return None
 
-            # Construct full URL using environment utilities
-            full_url = build_mcp_server_url(server_name)
+            # Check if a custom URL is provided
+            custom_url = self._extract_server_url(server_element)
 
-            return MCPServerConfig(mcp_server_name=name, mcp_server_unique_name=full_url)
+            # Determine the final URL: use custom URL if provided, otherwise construct it
+            final_url = custom_url if custom_url else build_mcp_server_url(server_name)
+
+            return MCPServerConfig(
+                mcp_server_name=name, mcp_server_unique_name=server_name, url=final_url
+            )
 
         except Exception:
             return None
@@ -425,7 +430,13 @@ def _parse_gateway_server_config(
             if not self._validate_server_strings(name, endpoint):
                 return None
 
-            return MCPServerConfig(mcp_server_name=name, mcp_server_unique_name=endpoint)
+            # Check if a custom URL is provided by the gateway
+            custom_url = self._extract_server_url(server_element)
+
+            # Determine the final URL: use custom URL if provided, otherwise use endpoint
+            final_url = custom_url if custom_url else endpoint
+
+            return MCPServerConfig(mcp_server_name=name, mcp_server_unique_name=endpoint, url=final_url)
 
         except Exception:
             return None
@@ -480,6 +491,21 @@ def _extract_server_unique_name(self, server_element: Dict[str, Any]) -> Optiona
             return server_element["mcpServerUniqueName"]
         return None
 
+    def _extract_server_url(self, server_element: Dict[str, Any]) -> Optional[str]:
+        """
+        Extracts custom server URL from configuration element.
+
+        Args:
+            server_element: Configuration dictionary.
+
+        Returns:
+            Server URL string or None.
+        """
+        # Check for 'url' field in both manifest and gateway responses
+        if "url" in server_element and isinstance(server_element["url"], str):
+            return server_element["url"]
+        return None
+
     def _validate_server_strings(self, name: Optional[str], unique_name: Optional[str]) -> bool:
         """
         Validates that server name and unique name are valid strings.

tests/tooling/__init__.py

@@ -0,0 +1,4 @@
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT License.
+
+"""Tests for tooling components."""

tests/tooling/test_mcp_server_configuration.py

@@ -0,0 +1,233 @@
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT License.
+
+"""Unit tests for MCP Server Configuration Service."""
+
+import json
+import os
+from pathlib import Path
+from typing import Dict, Any
+from unittest.mock import Mock, patch, AsyncMock, MagicMock
+import pytest
+import aiohttp
+
+from microsoft_agents_a365.tooling.services.mcp_tool_server_configuration_service import (
+    McpToolServerConfigurationService,
+)
+from microsoft_agents_a365.tooling.models import MCPServerConfig
+
+
+class TestMCPServerConfig:
+    """Tests for MCPServerConfig model."""
+
+    def test_mcp_server_config_with_custom_url(self):
+        """Test that MCPServerConfig can be created with a custom URL."""
+        config = MCPServerConfig(
+            mcp_server_name="TestServer",
+            mcp_server_unique_name="test_server",
+            url="https://custom.mcp.server/endpoint",
+        )
+
+        assert config.mcp_server_name == "TestServer"
+        assert config.mcp_server_unique_name == "test_server"
+        assert config.url == "https://custom.mcp.server/endpoint"
+
+    def test_mcp_server_config_without_custom_url(self):
+        """Test that MCPServerConfig works without a custom URL."""
+        config = MCPServerConfig(
+            mcp_server_name="TestServer",
+            mcp_server_unique_name="test_server",
+        )
+
+        assert config.mcp_server_name == "TestServer"
+        assert config.mcp_server_unique_name == "test_server"
+        assert config.url is None
+
+    def test_mcp_server_config_validation(self):
+        """Test that MCPServerConfig validates required fields."""
+        with pytest.raises(ValueError, match="mcp_server_name cannot be empty"):
+            MCPServerConfig(mcp_server_name="", mcp_server_unique_name="test")
+
+        with pytest.raises(ValueError, match="mcp_server_unique_name cannot be empty"):
+            MCPServerConfig(mcp_server_name="test", mcp_server_unique_name="")
+
+
+class TestMcpToolServerConfigurationService:
+    """Tests for McpToolServerConfigurationService."""
+
+    @pytest.fixture
+    def service(self):
+        """Create a service instance for testing."""
+        return McpToolServerConfigurationService()
+
+    @pytest.fixture
+    def mock_manifest_data(self) -> Dict[str, Any]:
+        """Create mock manifest data."""
+        return {
+            "mcpServers": [
+                {
+                    "mcpServerName": "TestServer1",
+                    "mcpServerUniqueName": "test_server_1",
+                },
+                {
+                    "mcpServerName": "TestServer2",
+                    "mcpServerUniqueName": "test_server_2",
+                    "url": "https://custom.server.com/mcp",
+                },
+            ]
+        }
+
+    def test_extract_server_url_from_manifest(self, service):
+        """Test extracting custom URL from manifest element."""
+        # Test with url field
+        element = {"url": "https://custom.url.com"}
+        url = service._extract_server_url(element)
+        assert url == "https://custom.url.com"
+
+        # Test with no URL
+        element = {}
+        url = service._extract_server_url(element)
+        assert url is None
+
+    def test_parse_manifest_server_config_with_custom_url(self, service):
+        """Test parsing manifest config with custom URL."""
+        server_element = {
+            "mcpServerName": "CustomServer",
+            "mcpServerUniqueName": "custom_server",
+            "url": "https://my.custom.server/mcp",
+        }
+
+        config = service._parse_manifest_server_config(server_element)
+
+        assert config is not None
+        assert config.mcp_server_name == "CustomServer"
+        assert config.mcp_server_unique_name == "custom_server"
+        assert config.url == "https://my.custom.server/mcp"
+
+    @patch("microsoft_agents_a365.tooling.services.mcp_tool_server_configuration_service.build_mcp_server_url")
+    def test_parse_manifest_server_config_without_custom_url(self, mock_build_url, service):
+        """Test parsing manifest config without custom URL constructs URL."""
+        mock_build_url.return_value = "https://default.server/agents/servers/test_server"
+
+        server_element = {
+            "mcpServerName": "DefaultServer",
+            "mcpServerUniqueName": "test_server",
+        }
+
+        config = service._parse_manifest_server_config(server_element)
+
+        assert config is not None
+        assert config.mcp_server_name == "DefaultServer"
+        assert config.mcp_server_unique_name == "test_server"
+        # Without a custom URL, build_mcp_server_url constructs the full URL and stores it in the url field
+        assert config.url == "https://default.server/agents/servers/test_server"
+        mock_build_url.assert_called_once_with("test_server")
+
+    def test_parse_gateway_server_config_with_custom_url(self, service):
+        """Test parsing gateway config with custom URL."""
+        server_element = {
+            "mcpServerName": "GatewayServer",
+            "mcpServerUniqueName": "gateway_server_endpoint",
+            "url": "https://gateway.custom.url/mcp",
+        }
+
+        config = service._parse_gateway_server_config(server_element)
+
+        assert config is not None
+        assert config.mcp_server_name == "GatewayServer"
+        assert config.mcp_server_unique_name == "gateway_server_endpoint"
+        assert config.url == "https://gateway.custom.url/mcp"
+
+    def test_parse_gateway_server_config_without_custom_url(self, service):
+        """Test parsing gateway config without custom URL."""
+        server_element = {
+            "mcpServerName": "GatewayServer",
+            "mcpServerUniqueName": "https://gateway.default/endpoint",
+        }
+
+        config = service._parse_gateway_server_config(server_element)
+
+        assert config is not None
+        assert config.mcp_server_name == "GatewayServer"
+        assert config.mcp_server_unique_name == "https://gateway.default/endpoint"
+        # Without a custom URL, the endpoint is used as the url
+        assert config.url == "https://gateway.default/endpoint"
+
+    @patch.dict(os.environ, {"ENVIRONMENT": "Development"})
+    def test_is_development_scenario(self, service):
+        """Test development scenario detection."""
+        assert service._is_development_scenario() is True
+
+    @patch.dict(os.environ, {"ENVIRONMENT": "Production"})
+    def test_is_production_scenario(self, service):
+        """Test production scenario detection."""
+        assert service._is_development_scenario() is False
+
+    @patch.object(McpToolServerConfigurationService, "_load_servers_from_manifest")
+    @patch.dict(os.environ, {"ENVIRONMENT": "Development"})
+    @pytest.mark.asyncio
+    async def test_list_tool_servers_development(self, mock_load_manifest, service):
+        """Test listing servers in development mode."""
+        mock_servers = [
+            MCPServerConfig(
+                mcp_server_name="DevServer",
+                mcp_server_unique_name="dev_server",
+                url="https://dev.server/mcp",
+            )
+        ]
+        mock_load_manifest.return_value = mock_servers
+
+        servers = await service.list_tool_servers(
+            agentic_app_id="test-app-id", auth_token="test-token"
+        )
+
+        assert servers == mock_servers
+        mock_load_manifest.assert_called_once()
+
+    @patch("microsoft_agents_a365.tooling.services.mcp_tool_server_configuration_service.get_tooling_gateway_for_digital_worker")
+    @patch.dict(os.environ, {"ENVIRONMENT": "Production"})
+    @pytest.mark.asyncio
+    async def test_list_tool_servers_production_with_custom_url(self, mock_gateway_url, service):
+        """Test listing servers in production mode with custom URL."""
+        mock_gateway_url.return_value = "https://gateway.test/agents/test-app-id/mcpServers"
+
+        # Mock aiohttp response
+        mock_response_data = {
+            "mcpServers": [
+                {
+                    "mcpServerName": "ProdServer",
+                    "mcpServerUniqueName": "prod_server",
+                    "url": "https://prod.custom.url/mcp",
+                }
+            ]
+        }
+
+        with patch("aiohttp.ClientSession") as mock_session_class:
+            # Create proper async context managers
+            mock_response = MagicMock()
+            mock_response.status = 200
+            mock_response.text = AsyncMock(return_value=json.dumps(mock_response_data))
+
+            # Create async context manager for response
+            mock_response_cm = MagicMock()
+            mock_response_cm.__aenter__ = AsyncMock(return_value=mock_response)
+            mock_response_cm.__aexit__ = AsyncMock(return_value=None)
+
+            # Create async context manager for session
+            mock_session = MagicMock()
+            mock_session.get = MagicMock(return_value=mock_response_cm)
+
+            mock_session_cm = MagicMock()
+            mock_session_cm.__aenter__ = AsyncMock(return_value=mock_session)
+            mock_session_cm.__aexit__ = AsyncMock(return_value=None)
+
+            mock_session_class.return_value = mock_session_cm
+
+            servers = await service.list_tool_servers(
+                agentic_app_id="test-app-id", auth_token="test-token"
+            )
+
+            assert len(servers) == 1
+            assert servers[0].mcp_server_name == "ProdServer"
+            assert servers[0].mcp_server_unique_name == "prod_server"
+            assert servers[0].url == "https://prod.custom.url/mcp"