Remove MCP protocol test button

src/admin/app.py

@@ -20,7 +20,6 @@
 from src.admin.blueprints.format_search import bp as format_search_bp
 from src.admin.blueprints.gam import gam_bp
 from src.admin.blueprints.inventory import inventory_bp
-from src.admin.blueprints.mcp_test import mcp_test_bp
 from src.admin.blueprints.operations import operations_bp
 from src.admin.blueprints.policy import policy_bp
 from src.admin.blueprints.principals import principals_bp
@@ -280,7 +279,7 @@ def fix_hardcoded_urls(response):
 
     # Register blueprints
     app.register_blueprint(public_bp)  # Public routes (no auth required) - MUST BE FIRST
-    app.register_blueprint(core_bp)  # Core routes (/, /health, /static, /mcp-test)
+    app.register_blueprint(core_bp)  # Core routes (/, /health, /static)
     app.register_blueprint(auth_bp)  # No url_prefix - auth routes are at root
     app.register_blueprint(tenant_management_settings_bp)  # Tenant management settings at /settings
     app.register_blueprint(tenants_bp, url_prefix="/tenant")
@@ -302,7 +301,6 @@ def fix_hardcoded_urls(response):
     app.register_blueprint(api_bp, url_prefix="/api")
     app.register_blueprint(format_search_bp)  # Format search API (/api/formats)
     app.register_blueprint(activity_stream_bp)  # SSE endpoints - Flask handles /admin via script_name from nginx proxy
-    app.register_blueprint(mcp_test_bp)
     app.register_blueprint(schemas_bp)  # JSON Schema validation service
     app.register_blueprint(workflows_bp, url_prefix="/tenant")  # Workflow approval and review
     # app.register_blueprint(tasks_bp)  # Tasks management - Disabled, tasks eliminated in favor of workflow system

src/admin/blueprints/api.py

@@ -1,11 +1,8 @@
 """API management blueprint."""
 
-import asyncio
 import logging
 from datetime import UTC, datetime, timedelta
 
-from fastmcp.client import Client
-from fastmcp.client.transports import StreamableHttpTransport
 from flask import Blueprint, jsonify, request
 from sqlalchemy import func, select, text
 
@@ -367,82 +364,6 @@ def gam_get_advertisers():
     return jsonify({"error": "Not yet implemented"}), 501
 
 
-@api_bp.route("/mcp-test/call", methods=["POST"])
-@require_auth(admin_only=True)
-@log_admin_action("mcp_test_call")
-def mcp_test_call():
-    """Execute MCP protocol test call.
-
-    This endpoint allows super admins to test MCP protocol calls
-    through the Admin UI's protocol test interface.
-    """
-    try:
-        data = request.json
-        server_url = data.get("server_url")
-        tool_name = data.get("tool")  # The template sends 'tool' not 'method'
-        params = data.get("params", {})
-        auth_token = data.get("access_token")  # The template sends 'access_token'
-
-        if not all([server_url, tool_name, auth_token]):
-            return (
-                jsonify({"success": False, "error": "Missing required fields: server_url, tool, and access_token"}),
-                400,
-            )
-
-        # Get tenant from token
-        with get_db_session() as db_session:
-            principal = db_session.scalars(select(Principal).filter_by(access_token=auth_token)).first()
-            if not principal:
-                return jsonify({"success": False, "error": "Invalid auth token"}), 401
-
-            tenant_id = principal.tenant_id
-
-        # Create MCP client with proper headers
-        headers = {"x-adcp-auth": auth_token, "x-adcp-tenant": tenant_id}
-
-        # Make async call to MCP server
-        async def call_mcp():
-            transport = StreamableHttpTransport(url=server_url, headers=headers)
-            client = Client(transport=transport)
-
-            async with client:
-                # Pass params directly as per MCP specification (no req wrapper)
-                tool_params = params
-                result = await client.call_tool(tool_name, tool_params)
-
-                # Extract the structured content from the result
-                if hasattr(result, "structured_content"):
-                    return result.structured_content
-                elif hasattr(result, "content"):
-                    if isinstance(result.content, list) and len(result.content) > 0:
-                        return result.content[0].text if hasattr(result.content[0], "text") else str(result.content[0])
-                    return result.content
-                else:
-                    return str(result)
-
-        # Run the async function
-        loop = asyncio.new_event_loop()
-        asyncio.set_event_loop(loop)
-        try:
-            result = loop.run_until_complete(call_mcp())
-            return jsonify({"success": True, "result": result})
-        except Exception as e:
-            logger.error(f"MCP call failed: {str(e)}")
-            return jsonify(
-                {
-                    "success": False,
-                    "error": str(e),
-                    "details": {"tool": tool_name, "server_url": server_url, "params": params},
-                }
-            )
-        finally:
-            loop.close()
-
-    except Exception as e:
-        logger.error(f"MCP test call error: {str(e)}")
-        return jsonify({"success": False, "error": str(e)}), 500
-
-
 @api_bp.route("/gam/test-connection", methods=["POST"])
 @require_auth()
 @log_admin_action("test_gam_connection")

src/admin/blueprints/core.py

@@ -16,8 +16,6 @@
 from src.core.database.models import Principal, Tenant
 from src.core.domain_config import (
     extract_subdomain_from_host,
-    get_a2a_server_url,
-    get_mcp_server_url,
     is_sales_agent_domain,
 )
 
@@ -331,66 +329,6 @@ def send_static(path):
     return send_from_directory("static", path)
 
 
-@core_bp.route("/mcp-test")
-@require_auth(admin_only=True)
-def mcp_test():
-    """MCP protocol testing interface for super admins."""
-    # Get all tenants and their principals
-    with get_db_session() as db_session:
-        # Get tenants
-        stmt = select(Tenant).filter_by(is_active=True).order_by(Tenant.name)
-        tenant_objs = db_session.scalars(stmt).all()
-        tenants = []
-        for tenant in tenant_objs:
-            tenants.append(
-                {
-                    "tenant_id": tenant.tenant_id,
-                    "name": tenant.name,
-                    "subdomain": tenant.subdomain,
-                }
-            )
-
-        # Get all principals with their tenant info
-        stmt = select(Principal).join(Tenant).where(Tenant.is_active).order_by(Tenant.name, Principal.name)
-        principal_objs = db_session.scalars(stmt).all()
-        principals = []
-        for principal in principal_objs:
-            # Get the tenant name via relationship or separate query
-            stmt = select(Tenant.name).filter_by(tenant_id=principal.tenant_id)
-            tenant_name = db_session.scalar(stmt)
-            principals.append(
-                {
-                    "principal_id": principal.principal_id,
-                    "name": principal.name,
-                    "tenant_id": principal.tenant_id,
-                    "access_token": principal.access_token,
-                    "tenant_name": tenant_name,
-                }
-            )
-
-    # Get server URLs - use production URLs if in production, otherwise localhost
-    if os.environ.get("PRODUCTION") == "true":
-        # In production, both servers are accessible at the configured domain
-        mcp_server_url = get_mcp_server_url()
-        a2a_server_url = get_a2a_server_url()
-    else:
-        # In development, use localhost with the configured ports from environment
-        # Default to common development ports if not set
-        mcp_port = int(os.environ.get("ADCP_SALES_PORT", 8080))
-        a2a_port = int(os.environ.get("A2A_PORT", 8091))
-        mcp_server_url = f"http://localhost:{mcp_port}/mcp"  # Remove trailing slash
-        a2a_server_url = f"http://localhost:{a2a_port}/a2a"
-
-    return render_template(
-        "mcp_test.html",
-        tenants=tenants,
-        principals=principals,
-        server_url=mcp_server_url,  # Keep legacy name for MCP server
-        mcp_server_url=mcp_server_url,
-        a2a_server_url=a2a_server_url,
-    )
-
-
 @core_bp.route("/admin/tenant/<tenant_id>/reactivate", methods=["POST"])
 @require_auth(admin_only=True)
 @log_admin_action("reactivate_tenant")

templates/index.html

@@ -8,7 +8,6 @@
         <h2>Accounts</h2>
         <div>
             {% if session.role == 'super_admin' %}
-            <a href="{{ script_name }}/mcp-test" class="btn btn-secondary">MCP Protocol Test</a>
             <a href="{{ script_name }}/settings" class="btn btn-secondary">Settings</a>
             {% endif %}
             <a href="{{ script_name }}/create_tenant" class="btn">Create New Account</a>