This example demonstrates how to use the UTCP WebSocket transport for real-time communication.
The WebSocket transport provides:
- Real-time bidirectional communication
- Tool discovery via WebSocket handshake
- Streaming tool execution
- Authentication support (API Key, Basic Auth, OAuth2)
- Automatic reconnection and keep-alive
websocket_server.py- Mock WebSocket server implementing UTCP protocolwebsocket_client.py- Client example using WebSocket transportproviders.json- WebSocket provider configuration
The UTCP WebSocket protocol uses JSON messages:
// Client sends:
{"type": "discover", "request_id": "unique_id"}
// Server responds:
{
"type": "discovery_response",
"request_id": "unique_id",
"tools": [...]
}// Client sends:
{
"type": "call_tool",
"request_id": "unique_id",
"tool_name": "tool_name",
"arguments": {...}
}
// Server responds:
{
"type": "tool_response",
"request_id": "unique_id",
"result": {...}
}- Start the mock WebSocket server:
python websocket_server.py- In another terminal, run the client:
python websocket_client.pyThe providers.json shows how to configure WebSocket providers with authentication:
[
{
"name": "websocket_tools",
"provider_type": "websocket",
"url": "ws://localhost:8765/ws",
"auth": {
"auth_type": "api_key",
"api_key": "your-api-key",
"var_name": "X-API-Key",
"location": "header"
},
"keep_alive": true,
"protocol": "utcp-v1"
}
]