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