forked from xiaohei/taiji-AI-PAD
35 lines
889 B
Python
35 lines
889 B
Python
"""Shared state container for the MCP server."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass, field
|
|
from typing import Dict, Optional, TYPE_CHECKING
|
|
|
|
import nats
|
|
import redis.asyncio as redis
|
|
|
|
from config import Settings
|
|
from mcp_protocol import MCPProtocolHandler
|
|
|
|
if TYPE_CHECKING: # pragma: no cover - typing helpers
|
|
from fastapi import WebSocket
|
|
|
|
|
|
@dataclass
|
|
class ServiceState:
|
|
"""Runtime resources shared across routes and background tasks."""
|
|
|
|
settings: Settings = field(default_factory=Settings)
|
|
redis_client: Optional[redis.Redis] = None
|
|
nats_client: Optional[nats.NATS] = None
|
|
mcp_handler: Optional[MCPProtocolHandler] = None
|
|
active_websockets: Dict[str, "WebSocket"] = field(default_factory=dict)
|
|
|
|
|
|
_state = ServiceState()
|
|
|
|
|
|
def get_state() -> ServiceState:
|
|
"""Expose the singleton service state."""
|
|
return _state
|