feat(core): add title field support and SSO ADFS integration

Add title column to tools, resources, and prompts tables following MCP
2025-11-25 spec precedence (title → annotations.title → name). Implement
_resolve_tool_title() helper with comprehensive doctests.

SSO enhancements:
- Add ADFS integration with tutorial documentation
- Expand SSO service test coverage (775+ new test lines)
- Add integration tests for ADFS flows

Performance improvements:
- Optimize gateway listings with eager-loaded tool counts
- Add selectinload for tools relationship in admin queries

Testing:
- Add unit tests for demo_a2a_agent.py script
- Expand gateway_service and sso_service test coverage
- Add ADFS integration test suite

Documentation:
- Add comprehensive ADFS SSO tutorial
- Update A2A agent documentation
- Remove duplicate config.schema.json files

Database:
- Migration a7f3c9e1b2d4: add title column (idempotent)
- Update schemas to include title field validation

Signed-off-by: Jonathan Springer <jps@s390x.com>

Author: jonpspri

.secrets.baseline

@@ -15,8 +15,8 @@
 from alembic import op
 import sqlalchemy as sa
 
-revision: str = "8f2e1c9b0d3a"
-down_revision: Union[str, Sequence[str], None] = "3f7e9d1a2b4c"
+revision: str = "8f2e1c9b0d3a"  # pragma: allowlist secret
+down_revision: Union[str, Sequence[str], None] = "3f7e9d1a2b4c"  # pragma: allowlist secret
 branch_labels: Union[str, Sequence[str], None] = None
 depends_on: Union[str, Sequence[str], None] = None
 

mcpgateway/alembic/versions/a2a_push_notification_configs_8f2e1c9b0d3a.py

@@ -17,8 +17,8 @@
 from alembic import op
 import sqlalchemy as sa
 
-revision: str = "3f7e9d1a2b4c"
-down_revision: Union[str, Sequence[str], None] = "a7f3c9e1b2d4"  # pragma: allowlist secret
+revision: str = "3f7e9d1a2b4c"  # pragma: allowlist secret
+down_revision: Union[str, Sequence[str], None] = "c2d3e4f5a6b7"  # pragma: allowlist secret
 branch_labels: Union[str, Sequence[str], None] = None
 depends_on: Union[str, Sequence[str], None] = None
 

mcpgateway/alembic/versions/a2a_v1_domain_models_3f7e9d1a2b4c.py

@@ -14,7 +14,7 @@
 import sqlalchemy as sa
 
 # revision identifiers, used by Alembic.
-revision: str = "a7f3c9e1b2d4"
+revision: str = "a7f3c9e1b2d4"  # pragma: allowlist secret
 down_revision: Union[str, Sequence[str], None] = "225bde88217e"
 branch_labels: Union[str, Sequence[str], None] = None
 depends_on: Union[str, Sequence[str], None] = None

mcpgateway/alembic/versions/a7f3c9e1b2d4_add_title_to_tools_resources_prompts.py

@@ -14,8 +14,8 @@
 import sqlalchemy as sa
 
 # revision identifiers, used by Alembic.
-revision: str = "ffe4494639d3"
-down_revision: Union[str, Sequence[str], None] = "8f2e1c9b0d3a"
+revision: str = "ffe4494639d3"  # pragma: allowlist secret
+down_revision: Union[str, Sequence[str], None] = "8f2e1c9b0d3a"  # pragma: allowlist secret
 branch_labels: Union[str, Sequence[str], None] = None
 depends_on: Union[str, Sequence[str], None] = None
 

mcpgateway/alembic/versions/ffe4494639d3_add_a2a_task_events_table.py

@@ -1257,6 +1257,8 @@ class Permissions:
     TOOLS_DELETE = "tools.delete"
     TOOLS_EXECUTE = "tools.execute"
 
+    TOOLS_MANAGE_PLUGINS = "tools.manage_plugins"
+
     # Resource permissions
     RESOURCES_CREATE = "resources.create"
     RESOURCES_READ = "resources.read"

mcpgateway/db.py

@@ -10,11 +10,16 @@
 from __future__ import annotations
 
 # Standard
+import binascii
 from dataclasses import dataclass
 import logging
 from typing import Any, Dict, Mapping, Optional
 import uuid
 
+# Third-Party
+from cryptography.exceptions import InvalidTag
+import orjson
+
 # First-Party
 from mcpgateway.utils.services_auth import decode_auth
 from mcpgateway.utils.url_auth import apply_query_param_auth, sanitize_url_for_logging
@@ -347,14 +352,36 @@ def prepare_a2a_invocation(
     if auth_type in {"basic", "bearer", "authheaders", "api_key"} and auth_value:
         if isinstance(auth_value, str):
             if auth_type == "api_key":
-                headers.setdefault("Authorization", f"Bearer {auth_value}")
+                # For api_key, try to decode from base64 first, but fall back to using raw value
+                try:
+                    decoded = decode_auth(auth_value)
+                    if isinstance(decoded, Mapping):
+                        # Extract the actual key value from the decoded dict
+                        api_key = next(iter(decoded.values())) if decoded else auth_value
+                        headers.setdefault("Authorization", f"Bearer {api_key}")
+                    else:
+                        # Fallback if decode returns a string directly
+                        headers.setdefault("Authorization", f"Bearer {decoded}")
+                except (InvalidTag, binascii.Error, orjson.JSONDecodeError, IndexError, ValueError):
+                    # If decoding fails (corrupted data, wrong key, invalid encoding, truncated input,
+                    # or invalid nonce/cipher parameters), use the raw value as the API key
+                    #
+                    # TODO:  Is this a logic failure?  Perhaps the gateway should
+                    # ensure all API Key style auths are encoded -- or vice versa
+                    #
+                    headers.setdefault("Authorization", f"Bearer {auth_value}")
             else:
                 decoded = decode_auth(auth_value)
                 if not isinstance(decoded, Mapping):
                     raise ValueError("Decoded A2A authentication payload must be a mapping")
                 headers.update({str(key): str(value) for key, value in decoded.items()})
         elif isinstance(auth_value, Mapping):
-            headers.update({str(key): str(value) for key, value in auth_value.items()})
+            if auth_type == "api_key":
+                # Extract the actual key value from the mapping
+                api_key = next(iter(auth_value.values()), "") if auth_value else ""
+                headers.setdefault("Authorization", f"Bearer {api_key}")
+            else:
+                headers.update({str(key): str(value) for key, value in auth_value.items()})
 
     auth_query_params_decrypted: Dict[str, str] = {}
     target_endpoint_url = endpoint_url

mcpgateway/services/a2a_protocol.py

@@ -1,4 +1,3 @@
-#!/usr/bin/env python3
 # -*- coding: utf-8 -*-
 """Integration tests for A2A agent support using an in-memory ASGI fixture.
 

tests/integration/test_a2a_sdk_integration.py

@@ -25,7 +25,6 @@
     prepare_a2a_invocation,
 )
 
-
 # ── is_v1_a2a_protocol ──────────────────────────────────────────────────────
 
 
@@ -552,7 +551,7 @@ def test_prepare_a2a_invocation_skips_query_param_decrypt_failures(monkeypatch):
         parameters={"query": "hello"},
         interaction_type="query",
         auth_type="query_param",
-        auth_query_params={"api_key": "bad"},
+        auth_query_params={"api_key": "bad"},  # pragma: allowlist secret
     )
 
     assert prepared.endpoint_url == "https://example.com/"
@@ -640,8 +639,8 @@ def test_prepare_a2a_invocation_rejects_non_mapping_decoded_auth(monkeypatch):
 
 def test_prepare_a2a_invocation_query_param_auth_applies(monkeypatch):
     """Successful query param auth decryption applies params to URL."""
-    monkeypatch.setattr("mcpgateway.services.a2a_protocol.decode_auth", lambda _val: {"api_key": "real-key"})
-    monkeypatch.setattr("mcpgateway.services.a2a_protocol.apply_query_param_auth", lambda url, params: url + "?api_key=real-key")
+    monkeypatch.setattr("mcpgateway.services.a2a_protocol.decode_auth", lambda _val: {"api_key": "real-key"})  # pragma: allowlist secret
+    monkeypatch.setattr("mcpgateway.services.a2a_protocol.apply_query_param_auth", lambda url, params: url + "?api_key=real-key")  # pragma: allowlist secret
 
     prepared = prepare_a2a_invocation(
         agent_type="generic",
@@ -650,7 +649,7 @@ def test_prepare_a2a_invocation_query_param_auth_applies(monkeypatch):
         parameters={"query": "hi"},
         interaction_type="query",
         auth_type="query_param",
-        auth_query_params={"api_key": "encrypted"},
+        auth_query_params={"api_key": "encrypted"},  # pragma: allowlist secret
     )
 
     assert "api_key=real-key" in prepared.endpoint_url
@@ -778,8 +777,8 @@ def test_prepare_a2a_invocation_v_prefixed_protocol_uses_v1_format():
 
 def test_prepare_a2a_invocation_preserves_encrypted_auth_fields(monkeypatch):
     """Query-param auth preserves encrypted blobs and sets base_endpoint_url."""
-    monkeypatch.setattr("mcpgateway.services.a2a_protocol.decode_auth", lambda _val: {"api_key": "decrypted-key"})
-    monkeypatch.setattr("mcpgateway.services.a2a_protocol.apply_query_param_auth", lambda url, params: url + "?api_key=decrypted-key")
+    monkeypatch.setattr("mcpgateway.services.a2a_protocol.decode_auth", lambda _val: {"api_key": "decrypted-key"})  # pragma: allowlist secret
+    monkeypatch.setattr("mcpgateway.services.a2a_protocol.apply_query_param_auth", lambda url, params: url + "?api_key=decrypted-key")  # pragma: allowlist secret
 
     prepared = prepare_a2a_invocation(
         agent_type="generic",
@@ -788,10 +787,10 @@ def test_prepare_a2a_invocation_preserves_encrypted_auth_fields(monkeypatch):
         parameters={"query": "hello"},
         interaction_type="query",
         auth_type="query_param",
-        auth_query_params={"api_key": "encrypted_blob"},
+        auth_query_params={"api_key": "encrypted_blob"},  # pragma: allowlist secret
     )
 
-    assert prepared.auth_query_params_encrypted == {"api_key": "encrypted_blob"}
+    assert prepared.auth_query_params_encrypted == {"api_key": "encrypted_blob"}  # pragma: allowlist secret
     assert prepared.base_endpoint_url == "https://example.com/"
 
 

tests/unit/mcpgateway/services/test_a2a_protocol.py

@@ -228,7 +228,7 @@ async def test_invoke_sends_encrypted_auth_fields(self):
             uses_jsonrpc=True,
             base_endpoint_url="https://agent.test/",
             auth_value_encrypted="enc-bearer-blob",
-            auth_query_params_encrypted={"api_key": "enc-qp-blob"},
+            auth_query_params_encrypted={"api_key": "enc-qp-blob"},  # pragma: allowlist secret
         )
 
         mock_response = MagicMock()
@@ -245,7 +245,7 @@ async def test_invoke_sends_encrypted_auth_fields(self):
         call_kwargs = mock_client.post.call_args
         payload = call_kwargs.kwargs["json"]
         assert payload["auth_headers_encrypted"] == "enc-bearer-blob"
-        assert payload["auth_query_params_encrypted"] == {"api_key": "enc-qp-blob"}
+        assert payload["auth_query_params_encrypted"] == {"api_key": "enc-qp-blob"}  # pragma: allowlist secret
 
     @pytest.mark.asyncio
     async def test_invoke_uses_base_endpoint_url_when_available(self):