Class MCPServer

src/codegraphcontext/server.py:181–870

The main MCP Server class. This class orchestrates all the major components of the application, including: - Database connection management (`DatabaseManager` or `FalkorDBManager`) - Background job tracking (`JobManager`) - File system watching for live updates (`CodeWatche

class MCPServer:
    """
    The main MCP Server class.
    
    This class orchestrates all the major components of the application, including:
    - Database connection management (`DatabaseManager` or `FalkorDBManager`)
    - Background job tracking (`JobManager`)
    - File system watching for live updates (`CodeWatcher`)
    - Tool handlers for graph building, code searching, etc.
    - The main JSON-RPC communication loop for interacting with an AI assistant.
    """

    def __init__(self, loop=None, cwd: Path | None = None):
        """
        Initializes the MCP server and its components. 
        
        Args:
            loop: The asyncio event loop to use. If not provided, it gets the current
                  running loop or creates a new one.
            cwd: Working directory used for context resolution. Defaults to Path.cwd().
        """
        self.cwd = (cwd or Path.cwd()).resolve()
        self.discovered_child_contexts: List[dict] = []
        self._context_note_pending = False
        self.disabled_tools: Set[str] = set()

        try:
            ctx = resolve_context(cwd=self.cwd)
            self.resolved_context = ctx

            if ctx.database and not os.environ.get('CGC_RUNTIME_DB_TYPE'):
                os.environ['CGC_RUNTIME_DB_TYPE'] = ctx.database

            self.db_manager = get_database_manager(db_path=ctx.db_path)
            self.db_manager.get_driver()

            if not ctx.is_local:
                try:
                    children = discover_child_contexts(self.cwd, max_depth=1)
                    if children:
                        self.discovered_child_contexts = [asdict(c) for c in children]
                        self._context_note_pending = True
                except Exception:
                    pass
        except ValueError as e:
            raise ValueError(f"Database configuration error: {e}")

        # Initialize managers for jobs and file watching.
        self.job_manager = JobManager()
        
        # Get the current event loop to pass to thread-sensitive components like the graph builder.
        if loop is None:
            try:
                loop = asyncio.get_running_loop()
            except RuntimeError:
                loop = asyncio.new_event_loop()
                asyncio.set_event_loop(loop)
        self.loop = loop