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.
(self, loop=None, cwd: Path | None = None)
| 191 | """ |
| 192 | |
| 193 | def __init__(self, loop=None, cwd: Path | None = None): |
| 194 | """ |
| 195 | Initializes the MCP server and its components. |
| 196 | |
| 197 | Args: |
| 198 | loop: The asyncio event loop to use. If not provided, it gets the current |
| 199 | running loop or creates a new one. |
| 200 | cwd: Working directory used for context resolution. Defaults to Path.cwd(). |
| 201 | """ |
| 202 | self.cwd = (cwd or Path.cwd()).resolve() |
| 203 | self.discovered_child_contexts: List[dict] = [] |
| 204 | self._context_note_pending = False |
| 205 | self.disabled_tools: Set[str] = set() |
| 206 | |
| 207 | try: |
| 208 | ctx = resolve_context(cwd=self.cwd) |
| 209 | self.resolved_context = ctx |
| 210 | |
| 211 | if ctx.database and not os.environ.get('CGC_RUNTIME_DB_TYPE'): |
| 212 | os.environ['CGC_RUNTIME_DB_TYPE'] = ctx.database |
| 213 | |
| 214 | self.db_manager = get_database_manager(db_path=ctx.db_path) |
| 215 | self.db_manager.get_driver() |
| 216 | |
| 217 | if not ctx.is_local: |
| 218 | try: |
| 219 | children = discover_child_contexts(self.cwd, max_depth=1) |
| 220 | if children: |
| 221 | self.discovered_child_contexts = [asdict(c) for c in children] |
| 222 | self._context_note_pending = True |
| 223 | except Exception: |
| 224 | pass |
| 225 | except ValueError as e: |
| 226 | raise ValueError(f"Database configuration error: {e}") |
| 227 | |
| 228 | # Initialize managers for jobs and file watching. |
| 229 | self.job_manager = JobManager() |
| 230 | |
| 231 | # Get the current event loop to pass to thread-sensitive components like the graph builder. |
| 232 | if loop is None: |
| 233 | try: |
| 234 | loop = asyncio.get_running_loop() |
| 235 | except RuntimeError: |
| 236 | loop = asyncio.new_event_loop() |
| 237 | asyncio.set_event_loop(loop) |
| 238 | self.loop = loop |
| 239 | |
| 240 | # Initialize all the tool handlers, passing them the necessary managers and the event loop. |
| 241 | self.graph_builder = GraphBuilder(self.db_manager, self.job_manager, loop) |
| 242 | self.code_finder = CodeFinder(self.db_manager) |
| 243 | self.code_watcher = CodeWatcher(self.graph_builder, self.job_manager) |
| 244 | |
| 245 | # Define the tool manifest that will be exposed to the AI assistant. |
| 246 | self._init_tools() |
| 247 | |
| 248 | def _init_tools(self): |
| 249 | """ |
nothing calls this directly
no test coverage detected