Add webhook capability and implement in streamable http

Author: pantanurag555

src/mcp/client/session.py

@@ -271,6 +271,7 @@ async def call_tool(
         arguments: dict[str, Any] | None = None,
         read_timeout_seconds: timedelta | None = None,
         progress_callback: ProgressFnT | None = None,
+        webhooks: list[types.Webhook] | None = None,
     ) -> types.CallToolResult:
         """Send a tools/call request with optional progress callback support."""
 
@@ -282,6 +283,7 @@ async def call_tool(
                         name=name,
                         arguments=arguments,
                     ),
+                    webhooks=webhooks,
                 )
             ),
             types.CallToolResult,

src/mcp/server/fastmcp/server.py

@@ -63,6 +63,7 @@
 from mcp.types import Resource as MCPResource
 from mcp.types import ResourceTemplate as MCPResourceTemplate
 from mcp.types import Tool as MCPTool
+from mcp.types import Webhook
 
 logger = get_logger(__name__)
 
@@ -99,6 +100,7 @@ class Settings(BaseSettings, Generic[LifespanResultT]):
     stateless_http: bool = (
         False  # If True, uses true stateless mode (new transport per request)
     )
+    webhooks_supported: bool = False
 
     # resource settings
     warn_on_duplicate_resources: bool = True
@@ -150,11 +152,10 @@ def __init__(
         self._mcp_server = MCPServer(
             name=name or "FastMCP",
             instructions=instructions,
-            lifespan=(
-                lifespan_wrapper(self, self.settings.lifespan)
-                if self.settings.lifespan
-                else default_lifespan
-            ),
+            lifespan=lifespan_wrapper(self, self.settings.lifespan)
+            if self.settings.lifespan
+            else default_lifespan,
+            webhooks_supported=self.settings.webhooks_supported,
         )
         self._tool_manager = ToolManager(
             tools=tools, warn_on_duplicate_tools=self.settings.warn_on_duplicate_tools
@@ -165,6 +166,7 @@ def __init__(
         self._prompt_manager = PromptManager(
             warn_on_duplicate_prompts=self.settings.warn_on_duplicate_prompts
         )
+
         if (self.settings.auth is not None) != (auth_server_provider is not None):
             # TODO: after we support separate authorization servers (see
             # https://github.com/modelcontextprotocol/modelcontextprotocol/pull/284)
@@ -272,11 +274,19 @@ def get_context(self) -> Context[ServerSession, object]:
         return Context(request_context=request_context, fastmcp=self)
 
     async def call_tool(
-        self, name: str, arguments: dict[str, Any]
+        self,
+        name: str,
+        arguments: dict[str, Any],
+        webhooks: list[Webhook] | None = None,
     ) -> Sequence[TextContent | ImageContent | EmbeddedResource]:
         """Call a tool by name with arguments."""
         context = self.get_context()
-        result = await self._tool_manager.call_tool(name, arguments, context=context)
+        result = await self._tool_manager.call_tool(
+            name,
+            arguments,
+            context=context,
+            webhooks=webhooks
+        )
         converted_result = _convert_to_content(result)
         return converted_result
 
@@ -777,6 +787,7 @@ def streamable_http_app(self) -> Starlette:
                 event_store=self._event_store,
                 json_response=self.settings.json_response,
                 stateless=self.settings.stateless_http,  # Use the stateless setting
+                webhooks_supported=self.settings.webhooks_supported,  # Use the webhooks supported setting
             )
 
         # Create the ASGI handler
@@ -929,6 +940,7 @@ def my_tool(x: int, ctx: Context) -> str:
 
     _request_context: RequestContext[ServerSessionT, LifespanContextT] | None
     _fastmcp: FastMCP | None
+    has_webhook: bool = False
 
     def __init__(
         self,

src/mcp/server/fastmcp/tools/tool_manager.py

@@ -7,7 +7,7 @@
 from mcp.server.fastmcp.tools.base import Tool
 from mcp.server.fastmcp.utilities.logging import get_logger
 from mcp.shared.context import LifespanContextT
-from mcp.types import ToolAnnotations
+from mcp.types import ToolAnnotations, Webhook
 
 if TYPE_CHECKING:
     from mcp.server.fastmcp.server import Context
@@ -66,10 +66,16 @@ async def call_tool(
         name: str,
         arguments: dict[str, Any],
         context: Context[ServerSessionT, LifespanContextT] | None = None,
+        webhooks: list[Webhook] | None = None,
     ) -> Any:
         """Call a tool by name with arguments."""
         tool = self.get_tool(name)
         if not tool:
             raise ToolError(f"Unknown tool: {name}")
 
+        if context:
+            context.has_webhook = (
+                webhooks is not None and len(webhooks) > 0
+            )
+
         return await tool.run(arguments, context=context)

src/mcp/server/lowlevel/server.py

@@ -132,6 +132,7 @@ def __init__(
         lifespan: Callable[
             [Server[LifespanResultT]], AbstractAsyncContextManager[LifespanResultT]
         ] = lifespan,
+        webhooks_supported: bool = False,
     ):
         self.name = name
         self.version = version
@@ -144,6 +145,7 @@ def __init__(
         }
         self.notification_handlers: dict[type, Callable[..., Awaitable[None]]] = {}
         self.notification_options = NotificationOptions()
+        self.webhooks_supported = webhooks_supported
         logger.debug(f"Initializing server '{name}'")
 
     def create_initialization_options(
@@ -199,7 +201,8 @@ def get_capabilities(
         # Set tool capabilities if handler exists
         if types.ListToolsRequest in self.request_handlers:
             tools_capability = types.ToolsCapability(
-                listChanged=notification_options.tools_changed
+                listChanged=notification_options.tools_changed,
+                webhooksSupported=self.webhooks_supported,
             )
 
         # Set logging capabilities if handler exists
@@ -409,7 +412,7 @@ def decorator(
 
             async def handler(req: types.CallToolRequest):
                 try:
-                    results = await func(req.params.name, (req.params.arguments or {}))
+                    results = await func(req.params.name, (req.params.arguments or {}), req.webhooks)
                     return types.ServerResult(
                         types.CallToolResult(content=list(results), isError=False)
                     )

src/mcp/server/streamable_http.py

@@ -7,6 +7,7 @@
 responses, with streaming support for long-running operations.
 """
 
+import asyncio
 import json
 import logging
 import re
@@ -24,6 +25,7 @@
 from starlette.responses import Response
 from starlette.types import Receive, Scope, Send
 
+from mcp.shared._httpx_utils import create_mcp_http_client
 from mcp.shared.message import ServerMessageMetadata, SessionMessage
 from mcp.types import (
     INTERNAL_ERROR,
@@ -36,6 +38,7 @@
     JSONRPCRequest,
     JSONRPCResponse,
     RequestId,
+    Webhook,
 )
 
 logger = logging.getLogger(__name__)
@@ -136,6 +139,7 @@ def __init__(
         self,
         mcp_session_id: str | None,
         is_json_response_enabled: bool = False,
+        is_webhooks_supported: bool = False,
         event_store: EventStore | None = None,
     ) -> None:
         """
@@ -146,6 +150,10 @@ def __init__(
                             Must contain only visible ASCII characters (0x21-0x7E).
             is_json_response_enabled: If True, return JSON responses for requests
                                     instead of SSE streams. Default is False.
+            is_webhooks_supported: If True and if webhooks are provided in
+                        tools/call request, the client will receive an Accepted
+                        HTTP response and the CallTool response will be sent to
+                        the webhook. Default is False.
             event_store: Event store for resumability support. If provided,
                         resumability will be enabled, allowing clients to
                         reconnect and resume messages.
@@ -162,6 +170,7 @@ def __init__(
 
         self.mcp_session_id = mcp_session_id
         self.is_json_response_enabled = is_json_response_enabled
+        self.is_webhooks_supported = is_webhooks_supported
         self._event_store = event_store
         self._request_streams: dict[
             RequestId,
@@ -410,9 +419,43 @@ async def _handle_post_request(
             ](0)
             request_stream_reader = self._request_streams[request_id][1]
 
+            session_message = SessionMessage(message)
+            if self._is_call_tool_request_with_webhooks(
+                session_message.message
+            ):
+                if self.is_webhooks_supported:
+                    response = self._create_json_response(
+                        JSONRPCMessage(root=JSONRPCResponse(
+                            jsonrpc="2.0",
+                            id=message.root.id,
+                            result={
+                                'content': [{
+                                    'type': 'text',
+                                    'text': 'Response will be forwarded to the webhook.'
+                                }],
+                                'isError': False
+                            },
+                        )),
+                        HTTPStatus.OK,
+                    )
+                    asyncio.create_task(
+                        self._send_response_to_webhooks(
+                            request_id, session_message, request_stream_reader
+                        )
+                    )
+                else:
+                    logger.exception("Webhooks not supported error")
+                    err = "Webhooks not supported"
+                    response = self._create_error_response(
+                        f"Validation error: {err}",
+                        HTTPStatus.BAD_REQUEST,
+                        INVALID_PARAMS,
+                    )
+                await response(scope, receive, send)
+                return
+
             if self.is_json_response_enabled:
                 # Process the message
-                session_message = SessionMessage(message)
                 await writer.send(session_message)
                 try:
                     # Process messages from the request-specific stream
@@ -531,6 +574,115 @@ async def sse_writer():
                 await writer.send(Exception(err))
             return
 
+
+    async def _send_response_to_webhooks(
+        self,
+        request_id: str,
+        session_message: SessionMessage,
+        request_stream_reader: MemoryObjectReceiveStream[EventMessage],
+    ):
+        webhooks: list[Webhook] = [Webhook(**webhook) for webhook in session_message.message.root.webhooks]
+        writer = self._read_stream_writer
+        if writer is None:
+            raise ValueError(
+                "No read stream writer available. Ensure connect() is called first."
+            )
+        await writer.send(session_message)
+
+        try:
+            response_message = JSONRPCError(
+                jsonrpc="2.0",
+                id="server-error",  # We don't have a request ID for general errors
+                error=ErrorData(
+                    code=INTERNAL_ERROR,
+                    message="Error processing request: No response received",
+                ),
+            )
+
+            if self.is_json_response_enabled:
+                # Process messages from the request-specific stream
+                # We need to collect all messages until we get a response
+                async for event_message in request_stream_reader:
+                    # If it's a response, this is what we're waiting for
+                    if isinstance(
+                        event_message.message.root, JSONRPCResponse | JSONRPCError
+                    ):
+                        response_message = event_message.message
+                        break
+                    # For notifications and request, keep waiting
+                    else:
+                        logger.debug(
+                            f"received: {event_message.message.root.method}"
+                        )        
+
+                await self._send_message_to_webhooks(webhooks, response_message)
+            else:
+                # Send each event on the request stream as a separate message
+                async for event_message in request_stream_reader:
+                    event_data = self._create_event_data(event_message)
+                    await self._send_message_to_webhooks(webhooks, event_data)
+
+                    # If response, remove from pending streams and close
+                    if isinstance(
+                        event_message.message.root,
+                        JSONRPCResponse | JSONRPCError,
+                    ):
+                        break
+
+        except Exception as e:
+            logger.exception(f"Error sending response to webhooks: {e}")
+
+        finally:
+            await self._clean_up_memory_streams(request_id)
+
+
+    async def _send_message_to_webhooks(
+        self,
+        webhooks: list[Webhook],
+        message: JSONRPCMessage | JSONRPCError | dict[str, str],
+    ):
+        for webhook in webhooks:
+            headers = {"Content-Type": CONTENT_TYPE_JSON}
+            # Add authorization headers
+            if webhook.authentication and webhook.authentication.credentials:
+                if webhook.authentication.strategy == "bearer":
+                    headers["Authorization"] = f"Bearer {webhook.authentication.credentials}"
+                elif webhook.authentication.strategy == "apiKey":
+                    headers["X-API-Key"] = webhook.authentication.credentials
+                elif webhook.authentication.strategy == "basic":
+                    try:
+                        # Try to parse as JSON
+                        creds_dict = json.loads(webhook.authentication.credentials)
+                        if "username" in creds_dict and "password" in creds_dict:
+                            # Create basic auth header from username and password
+                            import base64
+                            auth_string = f"{creds_dict['username']}:{creds_dict['password']}"
+                            credentials = base64.b64encode(auth_string.encode()).decode()
+                            headers["Authorization"] = f"Basic {credentials}"
+                    except:
+                        # Not JSON, use as-is
+                        headers["Authorization"] = f"Basic {webhook.authentication.credentials}"
+                elif webhook.authentication.strategy == "customHeader" and webhook.authentication.credentials:
+                    try:
+                        custom_headers = json.loads(webhook.authentication.credentials)
+                        headers.update(custom_headers)
+                    except:
+                        pass
+
+            async with create_mcp_http_client(headers=headers) as client:
+                try:
+                    if isinstance(message, JSONRPCMessage | JSONRPCError):
+                        await client.post(
+                            webhook.url,
+                            json=message.model_dump_json(by_alias=True, exclude_none=True),
+                        )
+                    else:
+                        await client.post(webhook.url, json=message)
+
+                except Exception as e:
+                    logger.exception(f"Error sending response to webhook {webhook.url}: {e}")
+
+
     async def _handle_get_request(self, request: Request, send: Send) -> None:
         """
         Handle GET request to establish SSE.
@@ -651,6 +803,18 @@ async def _handle_delete_request(self, request: Request, send: Send) -> None:
         )
         await response(request.scope, request.receive, send)
 
+
+    def _is_call_tool_request_with_webhooks(self, message: JSONRPCMessage) -> bool:
+        """Check if the request is a call tool request with webhooks."""
+        return (
+            isinstance(message.root, JSONRPCRequest)
+            and message.root.method == "tools/call"
+            and hasattr(message.root, "webhooks")
+            and message.root.webhooks is not None
+            and len(message.root.webhooks) > 0
+        )
+
+
     async def _terminate_session(self) -> None:
         """Terminate the current session, closing all streams.