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.

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 library, websockets. In the following example:

In the following example:

The MCP server provides tools for retrieving weather information.
WebSocket communication is implemented using fastapi.WebSocket.
MCP protocol support is implemented using the FastMCP library.
The base URL differs from the HTTP endpoint: it includes an additional /ws segment in the path. For your application, append your customer-defined path that matches the server-side route. In this example, the path is ws.

WebSocket MCP Example

import asyncio
from pathlib import Path
import sys
sys.dont_write_bytecode = True
from fastapi import FastAPI, WebSocket
from mcp.server import Server
from mcp.server.websocket import websocket_server
from mcp.types import Tool, TextContent
import uvicorn
import httpx
import json

# Add the src directory to Python path
src_path = str(Path(__file__).parent / "src")
sys.path.insert(0, src_path)

# Constants
NWS_API_BASE = "https://api.weather.gov"
USER_AGENT = "weather-app/1.0"

app = FastAPI()
weather_server = Server("weather")

async def make_nws_request(url: str) -> dict:
    """Make a request to the NWS API with proper error handling"""
    async with httpx.AsyncClient() as client:
        headers = {"User-Agent": USER_AGENT}
        response = await client.get(url, headers=headers)
        response.raise_for_status()
        return response.json()

@weather_server.list_tools()
async def handle_list_tools():
    """Handler for listing available tools"""
    return [
        Tool(
            name="get-weather",
            description="Get the current weather for a location",
            inputSchema={
                "type": "object",
                "required": ["latitude", "longitude"],
                "properties": {
                    "latitude": {"type": "number", "description": "Latitude of the location"},
                    "longitude": {"type": "number", "description": "Longitude of the location"}
                }
            }
        ),
        Tool(
            name="get-forecast",
            description="Get weather forecast for a location",
            inputSchema={
                "type": "object",
                "required": ["latitude", "longitude"],
                "properties": {
                    "latitude": {"type": "number", "description": "Latitude of the location"},
                    "longitude": {"type": "number", "description": "Longitude of the location"}
                }
            }
        )
    ]

@weather_server.call_tool()
async def handle_call_tool(name: str, arguments: dict | None) -> list[TextContent]:
    """Handler for calling tools"""
    if name == "get-weather":
        if not arguments or "latitude" not in arguments or "longitude" not in arguments:
            return [TextContent(type="text", text="Missing required arguments: latitude and longitude")]

        try:
            lat = float(arguments["latitude"])
            lon = float(arguments["longitude"])
            
            async with httpx.AsyncClient() as client:
                # First get the forecast grid endpoint
                points_url = f"{NWS_API_BASE}/points/{lat},{lon}"
                headers = {"User-Agent": USER_AGENT}
                
                response = await client.get(points_url, headers=headers)
                response.raise_for_status()
                
                data = response.json()
                forecast_url = data["properties"]["forecast"]
                
                # Now get the actual forecast
                response = await client.get(forecast_url, headers=headers)
                response.raise_for_status()
                
                forecast = response.json()
                current_period = forecast["properties"]["periods"][0]
                
                return [TextContent(
                    type="text",
                    text=f"Weather forecast for {current_period['name']}:\n"
                         f"Temperature: {current_period['temperature']}°{current_period['temperatureUnit']}\n"
                         f"Conditions: {current_period['shortForecast']}\n"
                         f"Wind: {current_period['windSpeed']} {current_period['windDirection']}"
                )]
                
        except Exception as e:
            return [TextContent(type="text", text=f"Error getting weather: {str(e)}")]
    
    elif name == "get-forecast":
        if not arguments or "latitude" not in arguments or "longitude" not in arguments:
            return [TextContent(type="text", text="Missing required arguments: latitude and longitude")]
        
        try:
            lat = float(arguments["latitude"])
            lon = float(arguments["longitude"])
            
            # First get the forecast grid endpoint
            points_url = f"{NWS_API_BASE}/points/{lat},{lon}"
            points_data = await make_nws_request(points_url)
            
            # Get the forecast URL from the points response
            forecast_url = points_data["properties"]["forecast"]
            forecast_data = await make_nws_request(forecast_url)
            
            # Extract relevant forecast information
            periods = forecast_data["properties"]["periods"]
            forecasts = [{
                "name": period["name"],
                "temperature": period["temperature"],
                "temperatureUnit": period["temperatureUnit"],
                "shortForecast": period["shortForecast"]
            } for period in periods]
            
            return [TextContent(type="text", text=json.dumps(forecasts, indent=2))]
        
        except Exception as e:
            return [TextContent(type="text", text=f"Error getting forecast: {str(e)}")]
    
    else:
        return [TextContent(type="text", text="Unknown tool")]

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    """WebSocket endpoint that handles client connections"""
    # Note: MCP subprotocol check removed for testing compatibility
    async with websocket_server(websocket.scope, websocket.receive, websocket.send) as (read_stream, write_stream):
        await weather_server.run(read_stream, write_stream, weather_server.create_initialization_options())

if __name__ == "__main__":
    uvicorn.run(app, host="127.0.0.1", port=8080, ws_per_message_deflate=False)

Oracle Cloud Infrastructure Documentation

Invoking Applications with WebSocket