Coverage for src/local_deep_research/mcp/server.py: 96%
328 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-09-06 15:42 +0000
1"""
2MCP Server for Local Deep Research.
4This module provides an MCP (Model Context Protocol) server that exposes
5LDR's research capabilities to AI agents like Claude.
7Security Notice:
8 This server is designed for LOCAL USE ONLY via STDIO transport
9 (e.g., Claude Desktop). It has no built-in authentication or rate
10 limiting. Do NOT expose this server over a network without implementing
11 proper security controls (OAuth, rate limiting, input validation).
13 When running locally via STDIO, security is provided by your operating
14 system's user permissions.
16Tools:
17 - quick_research: Fast research summary (1-5 min)
18 - detailed_research: Comprehensive analysis (5-15 min)
19 - generate_report: Full markdown report (10-30 min)
20 - analyze_documents: Search local document collection (30s-2 min)
21 - search: Raw search results without LLM processing (5-30s)
22 - list_search_engines: List available search engines
23 - list_strategies: List available research strategies
24 - get_configuration: Get current server configuration
26Usage:
27 python -m local_deep_research.mcp
28 # or
29 ldr-mcp
30"""
32import re
33import sys
34from collections.abc import Callable
35from typing import Any, Dict, Optional, cast
37from loguru import logger
38from mcp.server.fastmcp import FastMCP
40from local_deep_research.api.research_functions import (
41 analyze_documents as ldr_analyze_documents,
42 detailed_research as ldr_detailed_research,
43 generate_report as ldr_generate_report,
44 quick_summary as ldr_quick_summary,
45)
46from local_deep_research.api.settings_utils import create_settings_snapshot
47from local_deep_research.search_system_factory import (
48 get_available_strategies,
49)
50from ..utilities.type_utils import unwrap_setting
51from ..constants import DEFAULT_SEARCH_TOOL
52from ..security.egress.policy import PolicyDeniedError
54# Create FastMCP server instance
55mcp = FastMCP(
56 "local-deep-research",
57 instructions="AI-powered deep research assistant with iterative analysis using LLMs and web searches",
58)
60_HTTP_ERROR_CODE_RE = re.compile(r"(?<!\d)(?:401|404|429|503)(?!\d)")
63def _classify_error(error_msg: str) -> str:
64 """Classify error for client handling."""
65 error_lower = error_msg.lower()
66 # Exception messages can contain timestamps, counters, or OS thread IDs.
67 # Only standalone three-digit values are HTTP status codes; matching a
68 # substring inside a longer number makes classification nondeterministic.
69 status_codes = set(_HTTP_ERROR_CODE_RE.findall(error_msg))
70 if "503" in status_codes or "unavailable" in error_lower:
71 return "service_unavailable"
72 if "404" in status_codes or "not found" in error_lower:
73 return "model_not_found"
74 if (
75 "api key" in error_lower
76 or "authentication" in error_lower
77 or "unauthorized" in error_lower
78 or "401" in status_codes
79 ):
80 return "auth_error"
81 if "timeout" in error_lower or "timed out" in error_lower:
82 return "timeout"
83 if "rate limit" in error_lower or "429" in status_codes:
84 return "rate_limit"
85 if "connection" in error_lower:
86 return "connection_error"
87 if "validation" in error_lower or "invalid" in error_lower:
88 return "validation_error"
89 return "unknown"
92def _policy_denied_response(
93 error: PolicyDeniedError, operation: str
94) -> Dict[str, Any]:
95 """Build a machine-readable, leak-safe response for a PolicyDeniedError.
97 The PDP's short ``reason`` code (e.g. ``scope_mismatch_private_only``) is
98 safe to surface -- it is a machine code, never user content. The
99 ``target`` attribute is not (engine names / URLs may carry user content
100 or internal hostnames), so it is logged at audit level but never returned
101 to the client. Fail-closed is preserved by construction: this helper only
102 runs after the PEP has already raised, so the underlying engine run has
103 already been blocked.
104 """
105 reason = error.decision.reason
106 logger.bind(policy_audit=True).warning(
107 "MCP tool denied by egress policy",
108 operation=operation,
109 reason=reason,
110 )
111 return {
112 "status": "error",
113 "error_type": "policy_denied",
114 "reason": reason,
115 "error": f"{operation} denied by egress policy: {reason}.",
116 }
119class ValidationError(Exception):
120 """Raised when parameter validation fails."""
122 pass
125_COLLECTION_NAME_RE = re.compile(r"^[A-Za-z0-9 _-]{1,100}$")
128def _validate_range(
129 value: Any,
130 name: str,
131 min_val: int | float,
132 max_val: int | float,
133 *,
134 type_check: type | tuple[type, ...] = int,
135 allow_none: bool = True,
136 convert_to: Optional[Callable[[int | float], int | float]] = None,
137 type_error: Optional[str] = None,
138 min_error: Optional[str] = None,
139 max_error: Optional[str] = None,
140) -> int | float | None:
141 """Validate a numeric parameter and optionally normalize its type."""
142 if value is None:
143 if allow_none:
144 return None
145 raise ValidationError(type_error or f"{name} is required")
146 if not isinstance(value, type_check):
147 raise ValidationError(type_error or f"{name} must be a number")
148 numeric_value = cast(int | float, value)
149 if not numeric_value >= min_val:
150 raise ValidationError(min_error or f"{name} must be at least {min_val}")
151 if not numeric_value <= max_val:
152 raise ValidationError(max_error or f"{name} cannot exceed {max_val}")
153 return (
154 convert_to(numeric_value) if convert_to is not None else numeric_value
155 )
158def _error_result(
159 error: Exception | str,
160 *,
161 operation: Optional[str] = None,
162 error_type: Optional[str] = None,
163) -> Dict[str, Any]:
164 """Build an MCP error result without changing its public message shape.
166 ``operation`` is the complete operation-specific wording that precedes
167 ``(<error_type>)``. Omitting it preserves ``error`` verbatim, which is
168 used for validation and other deliberately client-visible messages.
169 """
170 error_text = str(error)
171 resolved_type = error_type or _classify_error(error_text)
172 public_message = (
173 error_text
174 if operation is None
175 else f"{operation} ({resolved_type}). Check server logs for details."
176 )
177 return {
178 "status": "error",
179 "error": public_message,
180 "error_type": resolved_type,
181 }
184def _validate_query(query: str) -> str:
185 """Validate and sanitize query parameter."""
186 if not query or not query.strip():
187 raise ValidationError("Query cannot be empty")
188 query = query.strip()
189 if len(query) > 10000:
190 raise ValidationError(
191 "Query exceeds maximum length of 10000 characters"
192 )
193 return query
196def _validate_iterations(
197 iterations: Optional[int], max_val: int = 20
198) -> Optional[int]:
199 """Validate iterations parameter."""
200 positive_error = "Iterations must be a positive integer"
201 return cast(
202 Optional[int],
203 _validate_range(
204 iterations,
205 "Iterations",
206 1,
207 max_val,
208 type_error=positive_error,
209 min_error=positive_error,
210 ),
211 )
214def _validate_questions_per_iteration(qpi: Optional[int]) -> Optional[int]:
215 """Validate questions_per_iteration parameter."""
216 positive_error = "Questions per iteration must be a positive integer"
217 return cast(
218 Optional[int],
219 _validate_range(
220 qpi,
221 "Questions per iteration",
222 1,
223 10,
224 type_error=positive_error,
225 min_error=positive_error,
226 ),
227 )
230def _validate_max_results(max_results: int) -> int:
231 """Validate max_results parameter."""
232 positive_error = "Max results must be a positive integer"
233 return cast(
234 int,
235 _validate_range(
236 max_results,
237 "Max results",
238 1,
239 100,
240 allow_none=False,
241 type_error=positive_error,
242 min_error=positive_error,
243 ),
244 )
247def _validate_searches_per_section(searches_per_section: int) -> int:
248 """Validate searches_per_section parameter."""
249 positive_error = "Searches per section must be a positive integer"
250 return cast(
251 int,
252 _validate_range(
253 searches_per_section,
254 "Searches per section",
255 1,
256 10,
257 allow_none=False,
258 type_error=positive_error,
259 min_error=positive_error,
260 ),
261 )
264def _validate_temperature(temperature: Optional[float]) -> Optional[float]:
265 """Validate and normalize a temperature setting."""
266 range_error = "Temperature must be between 0.0 and 2.0"
267 return cast(
268 Optional[float],
269 _validate_range(
270 temperature,
271 "Temperature",
272 0.0,
273 2.0,
274 type_check=(int, float),
275 convert_to=float,
276 type_error="Temperature must be a number",
277 min_error=range_error,
278 max_error=range_error,
279 ),
280 )
283def _validate_search_engine(engine: Optional[str]) -> Optional[str]:
284 """Validate search engine name against available engines."""
285 if engine is None:
286 return None
287 engine = engine.strip()
288 if not engine:
289 return None
290 try:
291 from local_deep_research.web_search_engines.search_engines_config import (
292 search_config,
293 )
295 settings = create_settings_snapshot()
296 available = search_config(settings_snapshot=settings)
297 if engine not in available:
298 available_names = sorted(available.keys())
299 raise ValidationError( # noqa: TRY301
300 f"Unknown search engine '{engine}'. Available: {', '.join(available_names)}"
301 )
302 except ValidationError:
303 raise
304 except Exception:
305 logger.exception("Could not load engine config to validate engine")
306 raise ValidationError(
307 f"Cannot validate search engine '{engine}': engine configuration unavailable"
308 )
309 return engine
312def _validate_strategy(strategy: Optional[str]) -> Optional[str]:
313 """Validate strategy name against available strategies."""
314 if strategy is None: 314 ↛ 315line 314 didn't jump to line 315 because the condition on line 314 was never true
315 return None
316 strategy = strategy.strip()
317 if not strategy:
318 return None
319 available = get_available_strategies()
320 available_names = [s["name"] for s in available]
321 if strategy not in available_names:
322 raise ValidationError(
323 f"Unknown strategy '{strategy}'. Available: {', '.join(available_names)}"
324 )
325 return strategy
328def _build_settings_overrides(
329 search_engine: Optional[str] = None,
330 strategy: Optional[str] = None,
331 iterations: Optional[int] = None,
332 questions_per_iteration: Optional[int] = None,
333 temperature: Optional[float] = None,
334) -> Dict[str, Any]:
335 """Build settings overrides dict from tool parameters."""
336 overrides: dict[str, Any] = {}
337 if search_engine is not None:
338 search_engine = _validate_search_engine(search_engine)
339 if search_engine:
340 overrides["search.tool"] = search_engine
341 if strategy is not None:
342 strategy = _validate_strategy(strategy)
343 if strategy:
344 overrides["search.search_strategy"] = strategy
345 if iterations is not None:
346 overrides["search.iterations"] = iterations
347 if questions_per_iteration is not None:
348 overrides["search.questions_per_iteration"] = questions_per_iteration
349 if temperature is not None:
350 overrides["llm.temperature"] = _validate_temperature(temperature)
351 return overrides
354# =============================================================================
355# Research Tools
356# =============================================================================
359@mcp.tool()
360def quick_research(
361 query: str,
362 search_engine: Optional[str] = None,
363 strategy: Optional[str] = None,
364 iterations: Optional[int] = None,
365 questions_per_iteration: Optional[int] = None,
366) -> Dict[str, Any]:
367 """
368 Perform quick research on a topic.
370 This tool performs a fast research summary on the given query. It searches
371 the web, analyzes sources, and generates a concise summary with findings.
373 IMPORTANT: This is a synchronous operation that typically takes 1-5 minutes
374 to complete depending on the complexity and configuration.
376 Args:
377 query: The research question or topic to investigate.
378 search_engine: Search engine to use (e.g., "wikipedia", "arxiv", "searxng").
379 Use list_search_engines() to see available options.
380 strategy: Research strategy to use (e.g., "source-based", "rapid", "iterative").
381 Use list_strategies() to see available options.
382 iterations: Number of search iterations (1-10). More iterations = deeper research.
383 questions_per_iteration: Questions to generate per iteration (1-5).
385 Returns:
386 Dictionary containing:
387 - status: "success" or "error"
388 - summary: The research summary text
389 - findings: List of detailed findings from each search
390 - sources: List of source URLs discovered
391 - iterations: Number of iterations performed
392 - error: Error message (only if status is "error")
393 - error_type: Error classification (only if status is "error")
394 """
395 try:
396 # Validate parameters
397 query = _validate_query(query)
398 iterations = _validate_iterations(iterations, max_val=10)
399 questions_per_iteration = _validate_questions_per_iteration(
400 questions_per_iteration
401 )
403 logger.info(f"Starting quick research for query: {query[:100]}...")
405 overrides = _build_settings_overrides(
406 search_engine=search_engine,
407 strategy=strategy,
408 iterations=iterations,
409 questions_per_iteration=questions_per_iteration,
410 )
412 settings = (
413 create_settings_snapshot(overrides=overrides)
414 if overrides
415 else create_settings_snapshot()
416 )
418 result = ldr_quick_summary(query, settings_snapshot=settings)
420 return {
421 "status": "success",
422 "summary": result.get("summary", ""),
423 "findings": result.get("findings", []),
424 "sources": result.get("sources", []),
425 "iterations": result.get("iterations", 0),
426 "formatted_findings": result.get("formatted_findings", ""),
427 }
429 except ValidationError as e:
430 logger.warning("Validation failed for quick research")
431 return _error_result(e, error_type="validation_error")
432 except PolicyDeniedError as e:
433 return _policy_denied_response(e, "Quick research")
434 except Exception as e:
435 logger.exception(
436 f"Quick research failed for query: {query[:100] if query else 'empty'}"
437 )
438 return _error_result(e, operation="Quick research failed")
441@mcp.tool()
442def detailed_research(
443 query: str,
444 search_engine: Optional[str] = None,
445 strategy: Optional[str] = None,
446 iterations: Optional[int] = None,
447 questions_per_iteration: Optional[int] = None,
448) -> Dict[str, Any]:
449 """
450 Perform detailed research with comprehensive analysis.
452 This tool performs a thorough research analysis on the given query, returning
453 structured data with detailed findings, sources, and metadata.
455 IMPORTANT: This is a synchronous operation that typically takes 5-15 minutes
456 to complete depending on the complexity and configuration.
458 Args:
459 query: The research question or topic to investigate.
460 search_engine: Search engine to use (e.g., "wikipedia", "arxiv", "searxng").
461 strategy: Research strategy to use (e.g., "source-based", "iterative", "evidence").
462 iterations: Number of search iterations (1-10). More iterations = deeper research.
463 questions_per_iteration: Questions to generate per iteration (1-5).
465 Returns:
466 Dictionary containing:
467 - status: "success" or "error"
468 - query: The original query
469 - research_id: Unique identifier for this research
470 - summary: The research summary text
471 - findings: List of detailed findings
472 - sources: List of source URLs
473 - iterations: Number of iterations performed
474 - metadata: Additional metadata (timestamp, search_tool, strategy)
475 - error/error_type: Error info (only if status is "error")
476 """
477 try:
478 # Validate parameters
479 query = _validate_query(query)
480 iterations = _validate_iterations(iterations, max_val=20)
481 questions_per_iteration = _validate_questions_per_iteration(
482 questions_per_iteration
483 )
485 logger.info(f"Starting detailed research for query: {query[:100]}...")
487 overrides = _build_settings_overrides(
488 search_engine=search_engine,
489 strategy=strategy,
490 iterations=iterations,
491 questions_per_iteration=questions_per_iteration,
492 )
494 settings = (
495 create_settings_snapshot(overrides=overrides)
496 if overrides
497 else create_settings_snapshot()
498 )
500 result = ldr_detailed_research(query, settings_snapshot=settings)
502 return {
503 "status": "success",
504 "query": result.get("query", query),
505 "research_id": result.get("research_id", ""),
506 "summary": result.get("summary", ""),
507 "findings": result.get("findings", []),
508 "sources": result.get("sources", []),
509 "iterations": result.get("iterations", 0),
510 "formatted_findings": result.get("formatted_findings", ""),
511 "metadata": result.get("metadata", {}),
512 }
514 except ValidationError as e:
515 logger.warning("Validation failed for detailed research")
516 return _error_result(e, error_type="validation_error")
517 except PolicyDeniedError as e:
518 return _policy_denied_response(e, "Detailed research")
519 except Exception as e:
520 logger.exception(
521 f"Detailed research failed for query: {query[:100] if query else 'empty'}"
522 )
523 return _error_result(e, operation="Detailed research failed")
526@mcp.tool()
527def generate_report(
528 query: str,
529 search_engine: Optional[str] = None,
530 searches_per_section: int = 2,
531) -> Dict[str, Any]:
532 """
533 Generate a comprehensive markdown research report.
535 This tool generates a full structured research report with sections,
536 citations, and comprehensive analysis. The output is formatted as markdown.
538 IMPORTANT: This is a synchronous operation that typically takes 10-30 minutes
539 to complete due to the comprehensive nature of the report.
541 Args:
542 query: The research question or topic for the report.
543 search_engine: Search engine to use (e.g., "wikipedia", "arxiv", "searxng").
544 searches_per_section: Number of searches per report section (1-10). Default is 2.
546 Returns:
547 Dictionary containing:
548 - status: "success" or "error"
549 - content: The full report content in markdown format
550 - metadata: Report metadata (timestamp, query)
551 - error/error_type: Error info (only if status is "error")
552 """
553 try:
554 # Validate parameters
555 query = _validate_query(query)
556 searches_per_section = _validate_searches_per_section(
557 searches_per_section
558 )
560 logger.info(f"Starting report generation for query: {query[:100]}...")
562 overrides = {}
563 if search_engine:
564 search_engine = _validate_search_engine(search_engine)
565 if search_engine: 565 ↛ 568line 565 didn't jump to line 568 because the condition on line 565 was always true
566 overrides["search.tool"] = search_engine
568 settings = (
569 create_settings_snapshot(overrides=overrides)
570 if overrides
571 else create_settings_snapshot()
572 )
574 result = ldr_generate_report(
575 query,
576 settings_snapshot=settings,
577 searches_per_section=searches_per_section,
578 )
580 return {
581 "status": "success",
582 "content": result.get("content", ""),
583 "metadata": result.get("metadata", {}),
584 }
586 except ValidationError as e:
587 logger.warning("Validation failed for report generation")
588 return _error_result(e, error_type="validation_error")
589 except PolicyDeniedError as e:
590 return _policy_denied_response(e, "Report generation")
591 except Exception as e:
592 logger.exception(
593 f"Report generation failed for query: {query[:100] if query else 'empty'}"
594 )
595 return _error_result(e, operation="Report generation failed")
598@mcp.tool()
599def analyze_documents(
600 query: str,
601 collection_name: str,
602 max_results: int = 10,
603) -> Dict[str, Any]:
604 """
605 Search and analyze documents in a local collection.
607 This tool performs RAG (Retrieval Augmented Generation) search on a
608 local document collection and generates a summary of relevant findings.
610 Args:
611 query: The search query for the documents.
612 collection_name: Name of the local document collection to search.
613 max_results: Maximum number of documents to retrieve (1-100). Default is 10.
615 Returns:
616 Dictionary containing:
617 - status: "success" or "error"
618 - summary: Summary of findings from the documents
619 - documents: List of matching documents with content and metadata
620 - collection: Name of the collection searched
621 - document_count: Number of documents found
622 - error/error_type: Error info (only if status is "error")
623 """
624 try:
625 # Validate parameters
626 query = _validate_query(query)
627 if not collection_name or not collection_name.strip():
628 raise ValidationError("Collection name cannot be empty") # noqa: TRY301
629 collection_name = collection_name.strip()
630 if not _COLLECTION_NAME_RE.match(collection_name):
631 raise ValidationError( # noqa: TRY301
632 "Collection name may only contain letters, digits, spaces, hyphens, and underscores (max 100 chars)"
633 )
634 max_results = _validate_max_results(max_results)
636 logger.info(
637 f"Analyzing documents in '{collection_name}' for query: {query[:100]}..."
638 )
640 # Build a settings snapshot the same way the other MCP tools do.
641 # Without this, analyze_documents falls back to JSON defaults +
642 # LDR_* env vars and silently ignores user-configured providers,
643 # API keys, and embedding model. Mirrors quick_research (line 278).
644 settings = create_settings_snapshot()
646 result = ldr_analyze_documents(
647 query=query,
648 collection_name=collection_name,
649 max_results=max_results,
650 settings_snapshot=settings,
651 )
653 return {
654 "status": "success",
655 "summary": result.get("summary", ""),
656 "documents": result.get("documents", []),
657 "collection": result.get("collection", collection_name),
658 "document_count": result.get("document_count", 0),
659 }
661 except ValidationError as e:
662 logger.warning("Validation failed for document analysis")
663 return _error_result(e, error_type="validation_error")
664 except PolicyDeniedError as e:
665 return _policy_denied_response(e, "Document analysis")
666 except Exception as e:
667 logger.exception(
668 f"Document analysis failed for collection: {collection_name if collection_name else 'empty'}"
669 )
670 return _error_result(e, operation="Document analysis failed")
673@mcp.tool()
674def search(
675 query: str,
676 engine: str,
677 max_results: int = 10,
678) -> Dict[str, Any]:
679 """
680 Search using a specific engine and return raw results without LLM processing.
682 This tool performs a direct search query against the specified engine and
683 returns raw results (title, link, snippet). No LLM is involved, making it
684 fast and free of LLM costs.
686 IMPORTANT: This is a fast operation, typically completing in 5-30 seconds.
688 Args:
689 query: The search query string.
690 engine: Search engine to use (e.g., "arxiv", "wikipedia", "searxng", "brave").
691 This is required — use list_search_engines() to see available options.
692 max_results: Maximum number of results to return (1-100). Default is 10.
694 Returns:
695 Dictionary containing:
696 - status: "success" or "error"
697 - query: The original query
698 - engine: The engine used
699 - result_count: Number of results returned
700 - results: List of results, each with title, link, and snippet
701 - error/error_type: Error info (only if status is "error")
702 """
703 try:
704 # Validate parameters
705 query = _validate_query(query)
706 max_results = _validate_max_results(max_results)
708 # Validate engine is non-empty (required parameter)
709 if not engine or not engine.strip(): 709 ↛ 710line 709 didn't jump to line 710 because the condition on line 709 was never true
710 raise ValidationError( # noqa: TRY301
711 "Engine name cannot be empty. Use list_search_engines() to see available options."
712 )
713 engine = engine.strip()
715 # Create settings snapshot (reused for all steps)
716 settings = create_settings_snapshot()
718 # Validate engine name against available engines
719 from local_deep_research.web_search_engines.search_engines_config import (
720 search_config,
721 )
723 engines_config = search_config(settings_snapshot=settings)
724 if engine not in engines_config:
725 available_names = sorted(engines_config.keys())
726 raise ValidationError( # noqa: TRY301
727 f"Unknown search engine '{engine}'. Available: {', '.join(available_names)}"
728 )
730 # Check API key requirement
731 engine_config = engines_config[engine]
732 if engine_config.get("requires_api_key", False):
733 api_key_setting = settings.get(
734 f"search.engine.web.{engine}.api_key"
735 )
736 api_key = None
737 if api_key_setting: 737 ↛ 738line 737 didn't jump to line 738 because the condition on line 737 was never true
738 api_key = (
739 api_key_setting.get("value")
740 if isinstance(api_key_setting, dict)
741 else api_key_setting
742 )
743 if not api_key: 743 ↛ 750line 743 didn't jump to line 750 because the condition on line 743 was always true
744 raise ValidationError( # noqa: TRY301
745 f"Engine '{engine}' requires an API key. "
746 f"Set the LDR_SEARCH_ENGINE_WEB_{engine.upper()}_API_KEY environment variable "
747 f"or configure it in the UI at search.engine.web.{engine}.api_key"
748 )
750 logger.info(
751 f"Starting search on '{engine}' for query: {query[:100]}..."
752 )
754 # Set thread-local settings context so that engine constructors
755 # which internally call get_llm() or get_setting_from_snapshot()
756 # (e.g., arxiv's JournalReputationFilter) can resolve settings.
757 from local_deep_research.config.thread_settings import (
758 clear_settings_context,
759 set_settings_context,
760 )
761 from local_deep_research.settings.manager import SnapshotSettingsContext
763 set_settings_context(SnapshotSettingsContext(settings))
764 try:
765 return _execute_search(query, engine, max_results, settings)
766 finally:
767 clear_settings_context()
769 except ValidationError as e:
770 logger.warning("Validation failed for search")
771 return _error_result(e, error_type="validation_error")
772 except PolicyDeniedError as e:
773 return _policy_denied_response(e, "Search")
774 except Exception as e:
775 logger.exception(
776 f"Search failed for query: {query[:100] if query else 'empty'}"
777 )
778 return _error_result(e, operation="Search failed")
781def _egress_audit_net(settings: Dict[str, Any]):
782 """Best-effort context manager that arms the egress audit-hook net for
783 a direct MCP search.
785 Direct MCP searches call ``engine.run()`` without going through
786 ``AdvancedSearchSystem`` (which arms the net itself), so under
787 PRIVATE_ONLY/STRICT the socket-level backstop would otherwise stay
788 inactive for this path. Returns a nullcontext when the policy cannot
789 be built or a context is already armed — the factory PEP remains the
790 primary enforcement, and an unevaluable policy must not break MCP.
791 """
792 from contextlib import nullcontext
794 try:
795 from local_deep_research.security.egress.audit_hook import (
796 active_egress_context,
797 get_active_context,
798 )
799 from local_deep_research.security.egress.policy import (
800 PolicyDeniedError,
801 context_from_snapshot,
802 )
803 except Exception:
804 return nullcontext()
806 if not settings or get_active_context() is not None:
807 return nullcontext()
808 try:
809 primary = unwrap_setting(
810 settings.get("search.tool", DEFAULT_SEARCH_TOOL)
811 )
812 ctx = context_from_snapshot(
813 settings,
814 primary or DEFAULT_SEARCH_TOOL,
815 username=settings.get("_username"),
816 )
817 except (PolicyDeniedError, ValueError):
818 logger.bind(policy_audit=True).debug(
819 "egress audit net not armed for MCP search: policy unevaluable",
820 exc_info=True,
821 )
822 return nullcontext()
823 except Exception:
824 return nullcontext()
825 return active_egress_context(ctx)
828def _execute_search(
829 query: str, engine: str, max_results: int, settings: Dict[str, Any]
830) -> Dict[str, Any]:
831 """Execute the search after settings context is established."""
832 from local_deep_research.web_search_engines.search_engine_factory import (
833 create_search_engine,
834 )
836 search_engine = create_search_engine(
837 engine_name=engine,
838 llm=None,
839 settings_snapshot=settings,
840 programmatic_mode=True,
841 max_results=max_results,
842 search_snippets_only=True,
843 )
845 if search_engine is None:
846 return _error_result(
847 f"Failed to create search engine '{engine}'. "
848 f"This engine may require an LLM or have other prerequisites. "
849 f"Check server logs for details.",
850 error_type="configuration_error",
851 )
853 try:
854 # Execute search with the egress audit-hook net armed (no-op
855 # under scopes that don't arm it or when policy is unavailable).
856 with _egress_audit_net(settings):
857 results = search_engine.run(query)
859 # Normalize results: ensure consistent 'snippet' key
860 for result in results:
861 if "snippet" not in result and "body" in result:
862 result["snippet"] = result["body"]
864 return {
865 "status": "success",
866 "query": query,
867 "engine": engine,
868 "result_count": len(results),
869 "results": results,
870 }
871 finally:
872 from local_deep_research.utilities.resource_utils import safe_close
874 safe_close(search_engine, "MCP search engine")
877# =============================================================================
878# Discovery Tools
879# =============================================================================
882@mcp.tool()
883def list_search_engines() -> Dict[str, Any]:
884 """
885 List available search engines.
887 Returns a list of search engines that can be used with the research tools.
888 Each engine has different strengths - some are better for academic research,
889 others for current events, etc.
891 Returns:
892 Dictionary containing:
893 - status: "success" or "error"
894 - engines: List of available search engine configurations
895 - error/error_type: Error info (only if status is "error")
896 """
897 try:
898 from local_deep_research.api.settings_utils import (
899 create_settings_snapshot,
900 )
901 from local_deep_research.web_search_engines.search_engines_config import (
902 search_config,
903 )
905 settings = create_settings_snapshot()
906 engines_config = search_config(settings_snapshot=settings)
908 engines = []
909 for name, config in engines_config.items():
910 engine_info = {
911 "name": name,
912 "description": config.get("description", ""),
913 "strengths": config.get("strengths", []),
914 "weaknesses": config.get("weaknesses", []),
915 "requires_api_key": config.get("requires_api_key", False),
916 "is_local": config.get("is_local", False),
917 }
918 engines.append(engine_info)
920 return {
921 "status": "success",
922 "engines": sorted(engines, key=lambda x: x["name"]),
923 }
925 except Exception as e:
926 logger.exception("Failed to list search engines")
927 return _error_result(e, operation="Failed to list search engines")
930@mcp.tool()
931def list_strategies() -> Dict[str, Any]:
932 """
933 List available research strategies.
935 Returns a list of research strategies that can be used with the research tools.
936 Each strategy has different characteristics suited for different types of queries.
938 Returns:
939 Dictionary containing:
940 - status: "success" or "error"
941 - strategies: List of available strategies with names and descriptions
942 - error/error_type: Error info (only if status is "error")
943 """
944 try:
945 return {
946 "status": "success",
947 "strategies": get_available_strategies(),
948 }
950 except Exception as e:
951 logger.exception("Failed to list strategies")
952 return _error_result(e, operation="Failed to list strategies")
955@mcp.tool()
956def get_configuration() -> Dict[str, Any]:
957 """
958 Get current server configuration.
960 Returns the current configuration settings being used by the MCP server,
961 including LLM provider, default search engine, and other settings.
963 Returns:
964 Dictionary containing:
965 - status: "success" or "error"
966 - config: Current configuration settings
967 - error/error_type: Error info (only if status is "error")
968 """
969 try:
970 from local_deep_research.api.settings_utils import (
971 create_settings_snapshot,
972 extract_setting_value,
973 )
975 settings = create_settings_snapshot()
977 config = {
978 "llm": {
979 "provider": extract_setting_value(
980 settings, "llm.provider", "unknown"
981 ),
982 "model": extract_setting_value(
983 settings, "llm.model", "unknown"
984 ),
985 "temperature": extract_setting_value(
986 settings, "llm.temperature", 0.7
987 ),
988 },
989 "search": {
990 "default_engine": extract_setting_value(
991 settings, "search.tool", DEFAULT_SEARCH_TOOL
992 ),
993 "default_strategy": extract_setting_value(
994 settings, "search.search_strategy", "source-based"
995 ),
996 "iterations": extract_setting_value(
997 settings, "search.iterations", 2
998 ),
999 "questions_per_iteration": extract_setting_value(
1000 settings, "search.questions_per_iteration", 3
1001 ),
1002 "max_results": extract_setting_value(
1003 settings, "search.max_results", 10
1004 ),
1005 },
1006 }
1008 return {
1009 "status": "success",
1010 "config": config,
1011 }
1013 except Exception as e:
1014 logger.exception("Failed to get configuration")
1015 return _error_result(e, operation="Failed to get configuration")
1018# =============================================================================
1019# Server Entry Point
1020# =============================================================================
1023def configure_mcp_logging(sink=None):
1024 """Send this process's logs to stderr, with control characters stripped.
1026 MCP uses stdout for JSON-RPC, so logging goes to stderr. This runs in a
1027 separate subprocess (ldr-mcp), so logger.remove() only affects this MCP
1028 process, not the main LDR application.
1029 """
1030 from ..security.log_sanitizer import sanitize_log_record
1032 logger.remove()
1033 # local_deep_research/__init__.py disables the package's own namespace, so
1034 # without this nothing logged from inside the package reaches the sink and
1035 # only third-party records get here. config_logger does the same on the
1036 # web path.
1037 logger.enable("local_deep_research")
1038 # The same patcher config_logger installs. Without it a query containing a
1039 # newline splits one WARNING into two stderr lines, and queries reach the
1040 # log verbatim (query[:80] in the citation-handler warnings). loguru holds
1041 # one patcher per process, so this has to be set here rather than
1042 # inherited.
1043 logger.configure(patcher=sanitize_log_record)
1044 # diagnose=False: loguru's default is True, which renders repr() of every
1045 # local in every traceback frame on exception. The many logger.exception()
1046 # call sites in this file run with frame locals that hold credentials
1047 # (api_key, Authorization headers, search-engine secrets), so leaving the
1048 # default on would write them to the MCP client's stderr log on any
1049 # failure. Companion to #4185 / config_logger's LDR_LOGURU_DIAGNOSE gate;
1050 # the MCP subprocess has no debug mode, so the gate is unconditionally off.
1051 #
1052 # enqueue=True: mirrors config_logger's stderr sink (utilities/log_utils.py).
1053 # Without it, this call blocks on stderr I/O while holding loguru's
1054 # handler lock, and when stderr back-pressures every logging thread piles
1055 # up behind that lock (#4431). The re-enable above (logger.enable(...))
1056 # means every INFO+ record from the whole research pipeline now reaches
1057 # this sink — that's exactly the condition #4431 was about, so this
1058 # sink needs the same protection. Note it's a bounded grace window
1059 # (roughly one queue pipe buffer, ~64KB on Linux), not immunity:
1060 # SimpleQueue.put() still runs inside emit()'s handler lock and blocks
1061 # once that buffer fills, so a stderr stall long enough to exhaust it
1062 # reproduces the same pile-up. Safe with the control-char patcher above
1063 # regardless: loguru applies core.patcher() to build the record and
1064 # formats the sink's output string before handing it to the enqueue
1065 # thread, so the message is already scrubbed by the time it's queued
1066 # (verified in loguru._logger.Logger._log / loguru._handler.Handler.emit).
1067 logger.add(
1068 sink if sink is not None else sys.stderr,
1069 level="INFO",
1070 format="{time:YYYY-MM-DD HH:mm:ss} | {level: <8} | {name}:{function}:{line} | {message}",
1071 diagnose=False,
1072 enqueue=True,
1073 )
1076def run_server():
1077 """Run the MCP server using STDIO transport."""
1078 configure_mcp_logging()
1080 # The drain `finally` opens immediately after the sink exists, not just
1081 # around `mcp.run()`: the startup window below logs through the same
1082 # enqueue=True sink, and the legacy-docstore sweep only catches
1083 # `Exception`. A KeyboardInterrupt/SystemExit raised inside
1084 # `migrate_legacy_docstores()` would otherwise escape with the startup
1085 # lines still queued and reproduce the very inversion this fixes.
1086 try:
1087 logger.info("Starting Local Deep Research MCP server...")
1089 # Phase-1 of the RAG plaintext-at-rest migration must run here too,
1090 # not only in the ldr-web entrypoint: an MCP-only deployment (Claude
1091 # Desktop / Code / OpenClaw) that never launches ldr-web would
1092 # otherwise never purge a pre-existing legacy plaintext <hash>.pkl
1093 # docstore, leaving chunk text on disk indefinitely. The sweep is
1094 # filesystem-only (no DB/login needed), idempotent (a clean tree
1095 # no-ops), and logs its own errors — wrapped so a cleanup issue never
1096 # blocks the server. Mirrors web/app.py's main().
1097 try:
1098 from ..vector_stores.legacy_cleanup import (
1099 migrate_legacy_docstores,
1100 )
1102 migrate_legacy_docstores()
1103 except Exception:
1104 logger.exception(
1105 "Legacy RAG docstore migration failed at MCP startup"
1106 )
1108 mcp.run(transport="stdio")
1109 finally:
1110 # Drain the enqueue=True stderr sink (see configure_mcp_logging)
1111 # before this function returns or raises, on every exit a
1112 # `finally` covers - normal disconnect, Exception, and BaseException
1113 # (KeyboardInterrupt) alike.
1114 #
1115 # loguru's enqueue=True handler runs a daemon writer thread reading
1116 # off a multiprocessing.SimpleQueue; loguru/__init__.py registers
1117 # `atexit.register(logger.remove)` for us, and draining (verified in
1118 # the installed loguru 0.7.3's _handler.py) is FIFO-sentinel based:
1119 # `Handler.stop()` enqueues a `None` sentinel and joins the writer
1120 # thread, so every message queued ahead of it is written first. On
1121 # its own, that atexit hook already keeps the *normal*-shutdown case
1122 # from losing buffered lines, as long as the interpreter reaches a
1123 # normal shutdown.
1124 #
1125 # What the atexit hook does NOT fix is ordering on an unhandled
1126 # exception. CPython's default excepthook prints the traceback to
1127 # stderr once the exception has finished propagating out of the top
1128 # frame, and only afterwards does the interpreter run its atexit
1129 # callbacks - so the atexit drain is strictly later than the
1130 # traceback. Without this `finally`, whatever is still sitting in
1131 # the queue when the exception is raised is written *after* the
1132 # traceback, so the client's stderr capture can read [traceback]
1133 # then [the log lines that led up to it], inverted from causal
1134 # order. (Only records still queued at raise time invert; anything
1135 # the writer thread already drained is unaffected.) Draining here,
1136 # in a `finally` around the server body, runs while the exception is
1137 # still propagating *through this frame* - strictly before it
1138 # reaches the top-level excepthook - so the log lines that explain
1139 # the crash are flushed first, then the traceback.
1140 #
1141 # This is `logger.complete()`, deliberately not `logger.remove()`
1142 # (verified in the installed loguru 0.7.3): `remove(None)` empties the
1143 # global handler table, so any record logged after this frame - a
1144 # caller catching the exception to log its own failure, or background
1145 # threads winding down - is silently dropped by `_log()`, which returns
1146 # early on `not core.handlers`. No caller does the former today: the
1147 # `ldr-mcp` console script (`pyproject.toml`:
1148 # `local_deep_research.mcp:run_server`) and `mcp/__main__.py`'s bare
1149 # `run_server()` call both invoke it with nothing wrapping the call to
1150 # catch and log a failure of its own. The property is kept as a general
1151 # guarantee `complete()` has over `remove()`, not a fix for an existing
1152 # caller. `complete()` drains the same queue (per enqueue handler it
1153 # enqueues a `True` confirmation token FIFO, then waits on the writer's
1154 # confirmation event) while *retaining* every handler, so those later
1155 # records still reach the sink. Sink teardown stays with its owner:
1156 # loguru's atexit-registered `logger.remove()`.
1157 #
1158 # Boundedness, precisely: the *enqueue* side is byte-bounded
1159 # back-pressure (SimpleQueue.put() runs under the handler lock and
1160 # stalls producers once the pipe buffer fills - the #4431 / #5804
1161 # window), but the drain itself is not time-bounded. The confirmation
1162 # wait in `Handler.complete_queue()` passes no timeout, so a sink that
1163 # blocks indefinitely blocks this `finally` indefinitely;
1164 # `Logger.complete()` additionally holds `_core.lock` across that wait,
1165 # so a concurrent `add`/`remove`/`configure` would block for the same
1166 # duration (ordinary logging would not - `_log()` never takes that
1167 # lock), and `tasks_to_complete()` briefly takes `_queue_lock`, the
1168 # same lock the writer thread holds around `sink.write()`. loguru
1169 # offers no honest bound to add here, so we accept the same
1170 # unbounded-in-time wait the writer thread already needs to drain its
1171 # queue. One practical consequence: a shutdown Ctrl-C now lands
1172 # *inside* this drain, so a second Ctrl-C raises KeyboardInterrupt out
1173 # of the `finally`, abandoning the drain and replacing whatever
1174 # exception was propagating.
1175 #
1176 # One case that leaves out: a *dead* writer thread. `complete_queue()`
1177 # first does `self._queue.put(True)` - which can itself block on a full
1178 # pipe before the wait even starts, while holding both
1179 # `Logger.complete()`'s `_core.lock` and its own `_confirmation_lock` -
1180 # then waits on `_confirmation_event` with no timeout and no check that
1181 # the writer thread is still alive to ever call `.set()` on it. If that
1182 # thread has already died, `logger.complete()` never returns.
1183 # `Handler.stop()` does not share this gap: `Thread.join()` returns
1184 # immediately once its target thread has already finished, so a dead
1185 # writer thread hangs `complete_queue()`, not `stop()`. Not reachable
1186 # with the shipped sink: it defaults to `sys.stderr`, which CPython
1187 # pins to `errors='backslashreplace'`, and a broken pipe raises
1188 # `OSError`, which `ErrorInterceptor.print()` (`_error_interceptor.py`)
1189 # itself catches around its own `sys.stderr.write()` when it reports a
1190 # sink failure - so the writer thread's loop keeps running. It becomes
1191 # reachable only if `sys.stderr` itself, not just this function's
1192 # `sink` argument, is ever pointed at a stream whose `write()` can
1193 # raise something other than `OSError`, since `ErrorInterceptor` always
1194 # reports to `sys.stderr` directly, independent of the handler's
1195 # configured sink. `Logger.enable()` and `.disable()` also take
1196 # `_core.lock` (`_logger.py`'s `_change_activation()`, ~line 1782), so
1197 # the full list of calls this drain can block is
1198 # `add`/`remove`/`configure`/`enable`/`disable`.
1199 #
1200 # Limit, not a regression: SIGTERM under its default disposition
1201 # kills the process outright and never runs this `finally` (nor
1202 # loguru's atexit hook), so anything still queued is lost - which is
1203 # what `scripts/mcp_smoke_test.sh`'s `timeout 5` does. Only in-band
1204 # exits (return, exception, KeyboardInterrupt) are covered.
1205 logger.complete()
1208if __name__ == "__main__": 1208 ↛ 1209line 1208 didn't jump to line 1209 because the condition on line 1208 was never true
1209 run_server()