Fix: Global Tools not listed for A2A Agents (Issue #841)

docs/docs/architecture/plugin-spec/.pages

@@ -14,4 +14,4 @@ nav:
   - Performance: 11-performance.md
   - Development: 12-development.md
   - Testing: 13-testing.md
-  - Conclusion: 14-conclusion.md
+  - Conclusion: 14-conclusion.md

docs/docs/architecture/plugin-spec/01-architecture.md

@@ -75,4 +75,3 @@ mcpgateway/plugins/framework/
 - Communicate via MCP protocol over various transports
 - 10-100ms latency depending on service and network
 - Examples: LlamaGuard, OpenAI Moderation, custom AI services
-

docs/docs/architecture/plugin-spec/02-core-components.md

@@ -3,7 +3,7 @@
 
 ### 3.1 Plugin Base Class
 
-The base plugin class, of which developers subclass and implement the hooks that are important for their plugins. Hook points are functions that appear interpose on existing MCP and agent-based functionality. 
+The base plugin class, of which developers subclass and implement the hooks that are important for their plugins. Hook points are functions that appear interpose on existing MCP and agent-based functionality.
 
 ```python
 class Plugin:
@@ -122,4 +122,3 @@ class PluginInstanceRegistry:
     async def shutdown(self) -> None:
         """Shutdown all registered plugins"""
 ```
-

docs/docs/architecture/plugin-spec/03-plugin-types.md

@@ -37,7 +37,7 @@ The configuration system supports both **native plugins** (running in-process) a
 
 ### 4.2 Plugin Configuration Schema
 
-Below is an example of a plugin configuration file.  A plugin configuration file can configure one or more plugins in a prioritized list as below.  Each individual plugin is an instance of the of a plugin class that subclasses the base `Plugin` object and implements a set of hooks as listed in the configuration.   
+Below is an example of a plugin configuration file.  A plugin configuration file can configure one or more plugins in a prioritized list as below.  Each individual plugin is an instance of the of a plugin class that subclasses the base `Plugin` object and implements a set of hooks as listed in the configuration.
 
 ```yaml
 # plugins/config.yaml
@@ -408,4 +408,3 @@ The manifest enables development tools to provide:
 - Follow established tag conventions within your organization
 
 The plugin manifest system provides a foundation for plugin ecosystem management, enabling better development workflows, automated tooling, and enhanced discoverability while maintaining consistency across plugin implementations.
-

docs/docs/architecture/plugin-spec/04-hook-architecture.md

@@ -186,7 +186,7 @@ return PluginResult(
 
 **Processing Model**:
 
-Plugin processing uses short circuiting to abort evaluation in the case of a violation and `continue_processing=False`. If the plugin needs to record side effects, such as the bookkeeping, these plugins should be executed first with the highest priority. 
+Plugin processing uses short circuiting to abort evaluation in the case of a violation and `continue_processing=False`. If the plugin needs to record side effects, such as the bookkeeping, these plugins should be executed first with the highest priority.
 
 ### 5.2 HTTP Header Hook Integration Example
 
@@ -477,4 +477,3 @@ async def process_elicitation_response(self, response: ElicitationResponse) -> b
 
     return True
 ```
-

docs/docs/architecture/plugin-spec/05-hook-system.md

@@ -184,4 +184,3 @@ This document covers administrative operation hooks:
 - Gateway Federation Hooks - Peer gateway management *(Future)*
 - A2A Agent Hooks - Agent-to-Agent integration management *(Future)*
 - Entity Lifecycle Hooks - Tool, resource, and prompt registration *(Future)*
-

docs/docs/architecture/plugin-spec/06-gateway-hooks.md

@@ -1667,4 +1667,3 @@ The gateway administrative hooks are organized into the following categories:
 - Implement proper timeout handling for elicitations
 - Cache frequently accessed data (permissions, quotas)
 - Use background tasks for non-critical operations
-

docs/docs/architecture/plugin-spec/07-security-hooks.md

@@ -758,4 +758,3 @@ async def resource_post_fetch(self, payload: ResourcePostFetchPayload, context:
 - Resource post-fetch may take longer due to content processing
 - Plugin execution is sequential within priority bands
 - Failed plugins don't affect other plugins (isolation)
-

docs/docs/architecture/plugin-spec/09-security.md

@@ -65,4 +65,3 @@ except Exception as e:
         raise PluginError(f"Plugin error: {plugin.name}")
     # Continue with next plugin in permissive mode
 ```
-

docs/docs/architecture/plugin-spec/10-error-handling.md

@@ -414,4 +414,3 @@ async def execute(self, plugins: list[PluginRef], ...) -> tuple[PluginResult[T],
                 raise PluginError(f"Plugin error: {plugin.name}")
             # Continue with next plugin
 ```
-

docs/docs/architecture/plugin-spec/11-performance.md

@@ -14,4 +14,3 @@
 - **Context management**: Handle 10,000+ concurrent request contexts
 - **Memory usage**: Base framework overhead <5MB
 - **Plugin loading**: Initialize plugins in <10 seconds
-

docs/docs/architecture/plugin-spec/12-development.md

@@ -286,4 +286,3 @@ class TestMyPlugin:
 - Include execution metrics
 - Provide health check endpoints
 - Support debugging modes
-

docs/docs/architecture/plugin-spec/13-testing.md

@@ -22,4 +22,3 @@ The plugin framework provides comprehensive testing support across multiple leve
 - Validate external plugin communication
 - Performance and load testing
 - Security validation
-

docs/docs/architecture/plugin-spec/14-conclusion.md

@@ -36,4 +36,3 @@ This specification defines a comprehensive, production-ready plugin framework fo
 
 This specification serves as the definitive guide for developing, deploying, and operating plugins within the MCP Context Forge ecosystem, ensuring consistency, security, and performance across all plugin implementations.
 **Document Version**: 1.0
-

docs/docs/architecture/plugin-spec/plugin-framework-specification.md

@@ -57,4 +57,3 @@ This specification covers:
 - **Plugin Manager**: Core service managing plugin lifecycle and execution
 - **Plugin Context**: Request-scoped state shared between plugins
 - **Plugin Configuration**: YAML-based plugin setup and parameters
-

mcpgateway/admin.py

@@ -8722,12 +8722,6 @@ async def admin_add_a2a_agent(
             federation_source=metadata["federation_source"],
         )
 
-        """
-        # Return redirect to admin page with A2A tab
-        root_path = request.scope.get("root_path", "")
-        return RedirectResponse(f"{root_path}/admin#a2a-agents", status_code=303)
-        """
-
         return JSONResponse(
             content={"message": "A2A agent created successfully!", "success": True},
             status_code=200,

mcpgateway/schemas.py

@@ -396,6 +396,7 @@ class ToolCreate(BaseModel):
     """
 
     model_config = ConfigDict(str_strip_whitespace=True, populate_by_name=True)
+    allow_auto: bool = False  # Internal flag to allow system-initiated A2A tool creation
 
     name: str = Field(..., description="Unique name for the tool")
     displayName: Optional[str] = Field(None, description="Display name for the tool (shown in UI)")  # noqa: N815
@@ -574,7 +575,7 @@ def validate_json_fields(cls, v: Dict[str, Any]) -> Dict[str, Any]:
     @field_validator("request_type")
     @classmethod
     def validate_request_type(cls, v: str, info: ValidationInfo) -> str:
-        """Validate request type based on integration type
+        """Validate request type based on integration type (REST, MCP, A2A)
 
         Args:
             v (str): Value to validate
@@ -587,42 +588,53 @@ def validate_request_type(cls, v: str, info: ValidationInfo) -> str:
             ValueError: When value is unsafe
 
         Examples:
-            >>> # Test REST integration types with valid method
             >>> from pydantic import ValidationInfo
-            >>> info = type('obj', (object,), {'data': {'integration_type': 'REST'}})
-            >>> ToolCreate.validate_request_type('POST', info)
+            >>> # REST integration types with valid methods
+            >>> info_rest = type('obj', (object,), {'data': {'integration_type': 'REST'}})
+            >>> ToolCreate.validate_request_type('POST', info_rest)
             'POST'
-
-            >>> # Test REST integration types
-            >>> info = type('obj', (object,), {'data': {'integration_type': 'REST'}})
-            >>> ToolCreate.validate_request_type('GET', info)
+            >>> ToolCreate.validate_request_type('GET', info_rest)
             'GET'
-
-            >>> # Test MCP integration types with valid transport
-            >>> info = type('obj', (object,), {'data': {'integration_type': 'MCP'}})
-            >>> ToolCreate.validate_request_type('SSE', info)
+            >>> # MCP integration types with valid transports
+            >>> info_mcp = type('obj', (object,), {'data': {'integration_type': 'MCP'}})
+            >>> ToolCreate.validate_request_type('SSE', info_mcp)
             'SSE'
-
-            >>> # Test invalid REST type
-            >>> info_rest = type('obj', (object,), {'data': {'integration_type': 'REST'}})
+            >>> ToolCreate.validate_request_type('STDIO', info_mcp)
+            'STDIO'
+            >>> # A2A integration type with valid method
+            >>> info_a2a = type('obj', (object,), {'data': {'integration_type': 'A2A'}})
+            >>> ToolCreate.validate_request_type('POST', info_a2a)
+            'POST'
+            >>> # Invalid REST type
             >>> try:
             ...     ToolCreate.validate_request_type('SSE', info_rest)
             ... except ValueError as e:
             ...     "not allowed for REST" in str(e)
             True
-
-            >>> # Test invalid integration type
-            >>> info = type('obj', (object,), {'data': {'integration_type': 'INVALID'}})
+            >>> # Invalid MCP type
+            >>> try:
+            ...     ToolCreate.validate_request_type('POST', info_mcp)
+            ... except ValueError as e:
+            ...     "not allowed for MCP" in str(e)
+            True
+            >>> # Invalid A2A type
+            >>> try:
+            ...     ToolCreate.validate_request_type('GET', info_a2a)
+            ... except ValueError as e:
+            ...     "not allowed for A2A" in str(e)
+            True
+            >>> # Invalid integration type
+            >>> info_invalid = type('obj', (object,), {'data': {'integration_type': 'INVALID'}})
             >>> try:
-            ...     ToolCreate.validate_request_type('GET', info)
+            ...     ToolCreate.validate_request_type('GET', info_invalid)
             ... except ValueError as e:
             ...     "Unknown integration type" in str(e)
             True
         """
 
         integration_type = info.data.get("integration_type")
 
-        if integration_type not in ["REST", "MCP"]:
+        if integration_type not in ["REST", "MCP", "A2A"]:
             raise ValueError(f"Unknown integration type: {integration_type}")
 
         if integration_type == "REST":
@@ -633,7 +645,10 @@ def validate_request_type(cls, v: str, info: ValidationInfo) -> str:
             allowed = ["SSE", "STDIO", "STREAMABLEHTTP"]
             if v not in allowed:
                 raise ValueError(f"Request type '{v}' not allowed for MCP. Only {allowed} transports are accepted.")
-
+        elif integration_type == "A2A":
+            allowed = ["POST"]
+            if v not in allowed:
+                raise ValueError(f"Request type '{v}' not allowed for A2A. Only {allowed} methods are accepted.")
         return v
 
     @model_validator(mode="before")
@@ -729,9 +744,10 @@ def prevent_manual_mcp_creation(cls, values: Dict[str, Any]) -> Dict[str, Any]:
             ValueError: If attempting to manually create MCP integration type
         """
         integration_type = values.get("integration_type")
+        allow_auto = values.get("allow_auto", False)
         if integration_type == "MCP":
             raise ValueError("Cannot manually create MCP tools. Add MCP servers via the Gateways interface - tools will be auto-discovered and registered with integration_type='MCP'.")
-        if integration_type == "A2A":
+        if integration_type == "A2A" and not allow_auto:
             raise ValueError("Cannot manually create A2A tools. Add A2A agents via the A2A interface - tools will be auto-created when agents are associated with servers.")
         return values
 

mcpgateway/services/a2a_service.py

@@ -26,6 +26,7 @@
 from mcpgateway.schemas import A2AAgentCreate, A2AAgentMetrics, A2AAgentRead, A2AAgentUpdate
 from mcpgateway.services.logging_service import LoggingService
 from mcpgateway.services.team_management_service import TeamManagementService
+from mcpgateway.services.tool_service import ToolService
 
 # Initialize logging service first
 logging_service = LoggingService()
@@ -205,6 +206,17 @@ async def register_agent(
         db.commit()
         db.refresh(new_agent)
 
+        # Automatically create a tool for the A2A agent if not already present
+        tool_service = ToolService()
+        await tool_service.create_tool_from_a2a_agent(
+            db=db,
+            agent=new_agent,
+            created_by=created_by,
+            created_from_ip=created_from_ip,
+            created_via=created_via,
+            created_user_agent=created_user_agent,
+        )
+
         logger.info(f"Registered new A2A agent: {new_agent.name} (ID: {new_agent.id})")
         return self._db_to_schema(new_agent)
 

mcpgateway/services/tool_service.py

@@ -807,6 +807,7 @@ async def invoke_tool(self, db: Session, name: str, arguments: Dict[str, Any], r
             True
         """
         # pylint: disable=comparison-with-callable
+        logger.info(f"Invoking tool: {name} with arguments: {arguments.keys() if arguments else None} and headers: {request_headers.keys() if request_headers else None}")
         tool = db.execute(select(DbTool).where(DbTool.name == name).where(DbTool.enabled)).scalar_one_or_none()
         if not tool:
             inactive_tool = db.execute(select(DbTool).where(DbTool.name == name).where(not_(DbTool.enabled))).scalar_one_or_none()
@@ -819,10 +820,9 @@ async def invoke_tool(self, db: Session, name: str, arguments: Dict[str, Any], r
 
         if not is_reachable:
             raise ToolNotFoundError(f"Tool '{name}' exists but is currently offline. Please verify if it is running.")
-
         # Check if this is an A2A tool and route to A2A service
         if tool.integration_type == "A2A" and tool.annotations and "a2a_agent_id" in tool.annotations:
-            return await self._invoke_a2a_tool(db, tool, arguments)
+            return await self._invoke_a2a_tool(db=db, tool=tool, arguments=arguments)
 
         # Plugin hook: tool pre-invoke
         context_table = None
@@ -1479,6 +1479,7 @@ async def create_tool_from_a2a_agent(
             ToolNameConflictError: If a tool with the same name already exists.
         """
         # Check if tool already exists for this agent
+        logger.info(f"testing Creating tool for A2A agent: {vars(agent)}")
         tool_name = f"a2a_{agent.slug}"
         existing_query = select(DbTool).where(DbTool.original_name == tool_name)
         existing_tool = db.execute(existing_query).scalar_one_or_none()
@@ -1503,6 +1504,7 @@ async def create_tool_from_a2a_agent(
                 },
                 "required": ["parameters"],
             },
+            allow_auto=True,
             annotations={
                 "title": f"A2A Agent: {agent.name}",
                 "a2a_agent_id": agent.id,
@@ -1536,6 +1538,7 @@ async def _invoke_a2a_tool(self, db: Session, tool: DbTool, arguments: Dict[str,
         Raises:
             ToolNotFoundError: If the A2A agent is not found.
         """
+
         # Extract A2A agent ID from tool annotations
         agent_id = tool.annotations.get("a2a_agent_id")
         if not agent_id:
@@ -1552,16 +1555,14 @@ async def _invoke_a2a_tool(self, db: Session, tool: DbTool, arguments: Dict[str,
             raise ToolNotFoundError(f"A2A agent '{agent.name}' is disabled")
 
         # Prepare parameters for A2A invocation
-        parameters = arguments.get("parameters", arguments)
-        interaction_type = arguments.get("interaction_type", "query")
 
         start_time = time.time()
         success = False
         error_message = None
 
         try:
             # Make the A2A agent call
-            response_data = await self._call_a2a_agent(agent, parameters, interaction_type)
+            response_data = await self._call_a2a_agent(agent, arguments)
             success = True
 
             # Convert A2A response to MCP ToolResult format
@@ -1594,28 +1595,44 @@ async def _invoke_a2a_tool(self, db: Session, tool: DbTool, arguments: Dict[str,
 
         return result
 
-    async def _call_a2a_agent(self, agent: DbA2AAgent, parameters: Dict[str, Any], interaction_type: str = "query") -> Dict[str, Any]:
+    async def _call_a2a_agent(self, agent: DbA2AAgent, parameters: Dict[str, Any]):
         """Call an A2A agent directly.
 
         Args:
             agent: The A2A agent to call.
             parameters: Parameters for the interaction.
-            interaction_type: Type of interaction.
 
         Returns:
             Response from the A2A agent.
 
         Raises:
             Exception: If the call fails.
         """
-        # Format request based on agent type and endpoint
-        if agent.agent_type in ["generic", "jsonrpc"] or agent.endpoint_url.endswith("/"):
-            # Use JSONRPC format for agents that expect it
-            request_data = {"jsonrpc": "2.0", "method": parameters.get("method", "message/send"), "params": parameters.get("params", parameters), "id": 1}
+        logger.info(f"Calling A2A agent '{agent.name}' at {agent.endpoint_url} with arguments: {parameters}")
+        # Patch: Build correct JSON-RPC params structure from flat UI input
+        params = None
+        # If UI sends flat fields, convert to nested message structure
+        if isinstance(parameters, dict) and "parameters" in parameters and "interaction_type" in parameters and isinstance(parameters["interaction_type"], str):
+            # Build the nested message object
+            message_id = f"admin-test-{int(time.time())}"
+            params = {"message": {"messageId": message_id, "role": "user", "parts": [{"type": "text", "text": parameters["interaction_type"]}]}}
+            method = parameters.get("parameters", "message/send")
         else:
-            # Use custom A2A format
-            request_data = {"interaction_type": interaction_type, "parameters": parameters, "protocol_version": agent.protocol_version}
+            # Already in correct format or unknown, pass through
+            params = parameters.get("params", parameters)
+            method = parameters.get("method", "message/send")
 
+        if agent.agent_type in ["generic", "jsonrpc"] or agent.endpoint_url.endswith("/"):
+            try:
+                request_data = {"jsonrpc": "2.0", "method": method, "params": params, "id": 1}
+                logger.info(f"invoke tool JSONRPC request_data prepared: {request_data}")
+            except Exception as e:
+                logger.error(f"Error preparing JSONRPC request data: {e}")
+                raise
+        else:
+            logger.info(f"invoke tool Using custom A2A format for A2A agent '{parameters}'")
+            request_data = {"interaction_type": parameters.get("interaction_type", "query"), "parameters": params, "protocol_version": agent.protocol_version}
+        logger.info(f"invoke tool request_data prepared: {request_data}")
         # Make HTTP request to the agent endpoint
         async with httpx.AsyncClient(timeout=30.0) as client:
             headers = {"Content-Type": "application/json"}

plugins/webhook_notification/README.md

@@ -219,4 +219,4 @@ The following variables are available in payload templates:
 Enable debug logging to see webhook delivery attempts:
 ```bash
 export LOG_LEVEL=DEBUG
-```
+```

plugins/webhook_notification/TESTING.md

@@ -427,4 +427,4 @@ jobs:
 }
 ```
 
-This comprehensive testing approach ensures the Webhook Notification Plugin is robust, reliable, and ready for production use.
+This comprehensive testing approach ensures the Webhook Notification Plugin is robust, reliable, and ready for production use.

plugins/webhook_notification/__init__.py

@@ -8,4 +8,4 @@
 
 from .webhook_notification import WebhookNotificationPlugin
 
-__all__ = ["WebhookNotificationPlugin"]
+__all__ = ["WebhookNotificationPlugin"]

plugins/webhook_notification/plugin-manifest.yaml

@@ -24,4 +24,4 @@ default_configs:
       "metadata": {{metadata}}
     }
   include_payload_data: false
-  max_payload_size: 1000
+  max_payload_size: 1000