Lesson 4 - Step 5 of 6

The Initialized Notification

Learn about the final step of the handshake where the client confirms it is ready.

Completing the Handshake

After the server responds to the initialize request, the client sends one more message: the notifications/initialized notification. This confirms the client has processed the server's capabilities and is ready for normal operation.

json

{
    "jsonrpc": "2.0",
    "method": "notifications/initialized"
}

This is a Notification

Notice there's no id field. This is a JSON-RPC notification, not a request. The server should NOT send a response. It's a one-way "I'm ready" signal.

The Complete Handshake Flow

Here's the full handshake sequence:

text

Client                                              Server
   │                                                    │
   │  1. initialize request                             │
   │    {jsonrpc, id:1, method:"initialize", params}    │
   │ ─────────────────────────────────────────────────► │
   │                                                    │
   │  2. initialize response                            │
   │    {jsonrpc, id:1, result: {capabilities, ...}}    │
   │ ◄───────────────────────────────────────────────── │
   │                                                    │
   │  3. initialized notification                       │
   │    {jsonrpc, method:"notifications/initialized"}   │
   │ ─────────────────────────────────────────────────► │
   │                                                    │
   │           ═══ HANDSHAKE COMPLETE ═══               │
   │                                                    │
   │  4. Normal operation begins                        │
   │    {jsonrpc, id:2, method:"tools/list"}            │
   │ ─────────────────────────────────────────────────► │
   │                                                    │

Why the Extra Step?

You might wonder why this notification is necessary. Several reasons:

Confirmation — The server knows the client received and processed its capabilities
Timing — The server can wait for this before accepting other requests
Setup opportunity — The client might need time to configure itself based on server capabilities

python

"""
Why initialized matters - a timing example.
"""

class MCPClient:
    def __init__(self):
        self.server_capabilities = None
        self.ready = False

    async def connect(self, server):
        # Step 1: Send initialize
        response = await server.send({
            "jsonrpc": "2.0",
            "id": 1,
            "method": "initialize",
            "params": {
                "protocolVersion": "2024-11-05",
                "capabilities": {},
                "clientInfo": {"name": "test", "version": "1.0"}
            }
        })

        # Step 2: Process server response
        self.server_capabilities = response["result"]["capabilities"]

        # Configure ourselves based on server capabilities
        if "tools" in self.server_capabilities:
            self.setup_tool_handling()
        if "resources" in self.server_capabilities:
            self.setup_resource_handling()

        # Step 3: NOW we're ready - tell the server
        await server.send({
            "jsonrpc": "2.0",
            "method": "notifications/initialized"
        })

        self.ready = True

    def setup_tool_handling(self):
        """Configure tool-related features."""
        print("Setting up tool UI elements...")

    def setup_resource_handling(self):
        """Configure resource-related features."""
        print("Setting up resource browser...")

Handling the Notification

Your server should handle this notification to transition to the ready state:

python

from enum import Enum

class ConnectionState(Enum):
    DISCONNECTED = "disconnected"
    INITIALIZING = "initializing"
    READY = "ready"

class MCPServer:
    def __init__(self):
        self.state = ConnectionState.DISCONNECTED
        self.client_info = None
        self.client_capabilities = None

    def handle_message(self, message: dict) -> dict | None:
        """Handle incoming JSON-RPC message.

        Returns response dict for requests, None for notifications.
        """
        method = message.get("method")
        is_notification = "id" not in message

        if method == "initialize":
            return self.handle_initialize(message)

        elif method == "notifications/initialized":
            self.handle_initialized_notification()
            return None  # No response for notifications!

        else:
            # All other methods require ready state
            if self.state != ConnectionState.READY:
                if is_notification:
                    return None  # Can't respond to notifications
                return {
                    "jsonrpc": "2.0",
                    "id": message.get("id"),
                    "error": {
                        "code": -32002,
                        "message": "Server not initialized"
                    }
                }

            return self.handle_request(message)

    def handle_initialize(self, message: dict) -> dict:
        """Handle initialize request."""
        if self.state != ConnectionState.DISCONNECTED:
            return {
                "jsonrpc": "2.0",
                "id": message.get("id"),
                "error": {
                    "code": -32600,
                    "message": "Already initialized"
                }
            }

        params = message.get("params", {})
        self.client_info = params.get("clientInfo", {})
        self.client_capabilities = params.get("capabilities", {})

        # Move to initializing state
        self.state = ConnectionState.INITIALIZING

        return {
            "jsonrpc": "2.0",
            "id": message.get("id"),
            "result": {
                "protocolVersion": "2024-11-05",
                "capabilities": {"tools": {}},
                "serverInfo": {"name": "my-server", "version": "1.0.0"}
            }
        }

    def handle_initialized_notification(self):
        """Handle initialized notification."""
        if self.state != ConnectionState.INITIALIZING:
            # Log warning but don't error - it's a notification
            print("Warning: Received initialized in unexpected state")
            return

        # Transition to ready!
        self.state = ConnectionState.READY
        print(f"Handshake complete! Client: {self.client_info.get('name')}")

    def handle_request(self, message: dict) -> dict:
        """Handle normal operational requests."""
        method = message.get("method")
        # ... route to appropriate handler
        return {
            "jsonrpc": "2.0",
            "id": message.get("id"),
            "result": {}
        }

State Transitions

Here's a state diagram for the handshake:

diagram

┌──────────────┐
│ DISCONNECTED │
└──────┬───────┘
       │
       │ Receive "initialize" request
       │ Send response
       ▼
┌──────────────┐
│ INITIALIZING │
└──────┬───────┘
       │
       │ Receive "notifications/initialized"
       │ (no response sent)
       ▼
┌──────────────┐
│    READY     │ ◄─── Normal operations
└──────────────┘

Error Cases

Handle these edge cases gracefully:

python

def handle_message(self, message: dict) -> dict | None:
    """Handle incoming message with edge case handling."""
    method = message.get("method")
    msg_id = message.get("id")  # None for notifications
    is_notification = msg_id is None

    # Case 1: Initialize when already initialized
    if method == "initialize" and self.state != ConnectionState.DISCONNECTED:
        return {
            "jsonrpc": "2.0",
            "id": msg_id,
            "error": {
                "code": -32600,
                "message": "Invalid Request: already initialized"
            }
        }

    # Case 2: Initialized notification at wrong time
    if method == "notifications/initialized":
        if self.state == ConnectionState.DISCONNECTED:
            # Got initialized before initialize - ignore it
            print("Warning: notifications/initialized before initialize")
            return None
        if self.state == ConnectionState.READY:
            # Already ready - ignore duplicate
            print("Warning: duplicate notifications/initialized")
            return None
        # Normal case - transition to ready
        self.state = ConnectionState.READY
        return None

    # Case 3: Other requests before ready
    if self.state != ConnectionState.READY and method != "initialize":
        if is_notification:
            print(f"Warning: notification {method} before ready")
            return None
        return {
            "jsonrpc": "2.0",
            "id": msg_id,
            "error": {
                "code": -32002,
                "message": "Server not initialized. Send initialize first."
            }
        }

    # Normal request handling
    return self.route_request(method, message)

Be Lenient with Notifications

Since notifications don't expect responses, be lenient when receiving them at unexpected times. Log a warning but don't crash. The client won't know if something went wrong anyway.

Testing the Handshake

Here's a simple test to verify your handshake handling:

python

def test_handshake():
    """Test the complete handshake sequence."""
    server = MCPServer()

    # Should start disconnected
    assert server.state == ConnectionState.DISCONNECTED

    # Step 1: Send initialize
    response = server.handle_message({
        "jsonrpc": "2.0",
        "id": 1,
        "method": "initialize",
        "params": {
            "protocolVersion": "2024-11-05",
            "capabilities": {},
            "clientInfo": {"name": "test", "version": "1.0"}
        }
    })

    # Should get success response
    assert "result" in response
    assert response["id"] == 1
    assert server.state == ConnectionState.INITIALIZING

    # Step 2: Try to call tools/list - should fail
    response = server.handle_message({
        "jsonrpc": "2.0",
        "id": 2,
        "method": "tools/list"
    })
    assert "error" in response
    assert response["error"]["code"] == -32002  # Not initialized

    # Step 3: Send initialized notification
    response = server.handle_message({
        "jsonrpc": "2.0",
        "method": "notifications/initialized"
    })

    # Should be None (no response to notifications)
    assert response is None
    assert server.state == ConnectionState.READY

    # Step 4: Now tools/list should work
    response = server.handle_message({
        "jsonrpc": "2.0",
        "id": 3,
        "method": "tools/list"
    })
    assert "result" in response

    print("Handshake test passed!")

if __name__ == "__main__":
    test_handshake()

Key Takeaways

notifications/initialized is a notification, not a request
The server must NOT respond to this message
It signals the handshake is complete and normal operation can begin
Track connection state to reject requests before initialization
Be lenient with notification timing - log warnings but don't crash