Invoking Applications with WebSocket

After you get an access token, you can invoke an OCI Generative AI application. This page describes how to use the access token with a WebSocket to invoke the client with a WebSocket MCP server example.

WebSocket is a communications protocol that provides full‑duplex, bidirectional communication over a single, long‑lived TCP connection. Unlike HTTP’s request/response model, it lets the server push data to the client as soon as it’s available (without polling), enabling low‑latency real‑time use cases such as chat, live dashboards, and multiplayer games.

You can use any WebSocket client library. The following example uses the Python websockets library. In the following example:

The base URL is different from HTTP endpoint: it includes an extra /ws segment in the path. For your application, you must also append your customer-defined path that matches the server-side route. In this example, the path is ws.
The OAuth token is provided in the Authorization header.
The example acts as an MCP client.

WebSocket Client Python Example

import websockets
import asyncio
import ssl

# Disable SSL certificate verification
ssl_context = ssl.create_default_context()
ssl_context.check_hostname = False
ssl_context.verify_mode = ssl.CERT_NONE

headers = {
    "Authorization": f'Bearer {token}',
    "Date": current_date,
    "ssl": False
}

url = f"wss://application.generativeai.ap-osaka-1.oci.oraclecloud.com/20251112/ws/hostedapplication/{hostedApplicationId}/ws"

async def main():
    try:
		async with websockets.connect(url, additional_headers=headers, ssl=ssl_context, subprotocols=[Subprotocol("mcp")]) as ws:
            print("Connected to WebSocket")

            # Initialize request
            init_request = {
                "jsonrpc": "2.0",
                "id": 1,
                "method": "initialize",
                "params": {
                    "protocolVersion": "2024-11-05",
                    "capabilities": {},
                    "clientInfo": {
                        "name": "http-test-client",
                        "version": "1.0"
                    }
                }
            }
            await ws.send(json.dumps(init_request))
            print("Sent: Initialization")

            result = await ws.recv()
            if isinstance(result, bytes):
                print("Received bytes:", result)
            else:
                print("Received text:", result)

            # List tools request
            list_tools_request = {
                "jsonrpc": "2.0",
                "id": 2,
                "method": "tools/list",
                "params": {}
            }
            await ws.send(json.dumps(list_tools_request))
            print("Sent: List tools")

            result = await ws.recv()
            if isinstance(result, bytes):
                print("Received bytes:", result)
            else:
                print("Received text:", result)

            # Call get-alerts tool request
            get_alerts_request = {
                "jsonrpc": "2.0",
                "id": 3,
                "method": "tools/call",
                "params": {
                    "name": "get-alerts",
                    "arguments": {
                        "state": "CA"
                    }
                }
            }
            await ws.send(json.dumps(get_alerts_request))
            print("Sent: call get alerts tool")
            result = await ws.recv()
            if isinstance(result, bytes):
                print("Received bytes:", result)
            else:
                print("Received text:", result)
    except Exception as e:
        print("Error:", e)

if __name__ == "__main__":
    asyncio.run(main())

Oracle Cloud Infrastructure Documentation

Invoking Applications with WebSocket