Coverage for src/local_deep_research/mcp/server.py: 94%
303 statements
« prev ^ index » next coverage.py v7.15.1, created at 2026-07-19 23:35 +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 typing import Any, Dict, Optional
36from loguru import logger
37from mcp.server.fastmcp import FastMCP
39from local_deep_research.api.research_functions import (
40 analyze_documents as ldr_analyze_documents,
41 detailed_research as ldr_detailed_research,
42 generate_report as ldr_generate_report,
43 quick_summary as ldr_quick_summary,
44)
45from local_deep_research.api.settings_utils import create_settings_snapshot
46from local_deep_research.search_system_factory import (
47 get_available_strategies,
48)
49from ..utilities.type_utils import unwrap_setting
50from ..constants import DEFAULT_SEARCH_TOOL
52# Create FastMCP server instance
53mcp = FastMCP(
54 "local-deep-research",
55 instructions="AI-powered deep research assistant with iterative analysis using LLMs and web searches",
56)
59def _classify_error(error_msg: str) -> str:
60 """Classify error for client handling."""
61 error_lower = error_msg.lower()
62 if "503" in error_msg or "unavailable" in error_lower:
63 return "service_unavailable"
64 if "404" in error_msg or "not found" in error_lower:
65 return "model_not_found"
66 if (
67 "api key" in error_lower
68 or "authentication" in error_lower
69 or "unauthorized" in error_lower
70 or "401" in error_msg
71 ):
72 return "auth_error"
73 if "timeout" in error_lower or "timed out" in error_lower:
74 return "timeout"
75 if "rate limit" in error_lower or "429" in error_msg:
76 return "rate_limit"
77 if "connection" in error_lower:
78 return "connection_error"
79 if "validation" in error_lower or "invalid" in error_lower:
80 return "validation_error"
81 return "unknown"
84class ValidationError(Exception):
85 """Raised when parameter validation fails."""
87 pass
90_COLLECTION_NAME_RE = re.compile(r"^[A-Za-z0-9 _-]{1,100}$")
93def _validate_query(query: str) -> str:
94 """Validate and sanitize query parameter."""
95 if not query or not query.strip():
96 raise ValidationError("Query cannot be empty")
97 query = query.strip()
98 if len(query) > 10000:
99 raise ValidationError(
100 "Query exceeds maximum length of 10000 characters"
101 )
102 return query
105def _validate_iterations(
106 iterations: Optional[int], max_val: int = 20
107) -> Optional[int]:
108 """Validate iterations parameter."""
109 if iterations is None:
110 return None
111 if not isinstance(iterations, int) or iterations < 1:
112 raise ValidationError("Iterations must be a positive integer")
113 if iterations > max_val:
114 raise ValidationError(f"Iterations cannot exceed {max_val}")
115 return iterations
118def _validate_questions_per_iteration(qpi: Optional[int]) -> Optional[int]:
119 """Validate questions_per_iteration parameter."""
120 if qpi is None:
121 return None
122 if not isinstance(qpi, int) or qpi < 1:
123 raise ValidationError(
124 "Questions per iteration must be a positive integer"
125 )
126 if qpi > 10:
127 raise ValidationError("Questions per iteration cannot exceed 10")
128 return qpi
131def _validate_max_results(max_results: int) -> int:
132 """Validate max_results parameter."""
133 if not isinstance(max_results, int) or max_results < 1:
134 raise ValidationError("Max results must be a positive integer")
135 if max_results > 100:
136 raise ValidationError("Max results cannot exceed 100")
137 return max_results
140def _validate_search_engine(engine: Optional[str]) -> Optional[str]:
141 """Validate search engine name against available engines."""
142 if engine is None:
143 return None
144 engine = engine.strip()
145 if not engine:
146 return None
147 try:
148 from local_deep_research.web_search_engines.search_engines_config import (
149 search_config,
150 )
152 settings = create_settings_snapshot()
153 available = search_config(settings_snapshot=settings)
154 if engine not in available:
155 available_names = sorted(available.keys())
156 raise ValidationError( # noqa: TRY301
157 f"Unknown search engine '{engine}'. Available: {', '.join(available_names)}"
158 )
159 except ValidationError:
160 raise
161 except Exception:
162 logger.exception("Could not load engine config to validate engine")
163 raise ValidationError(
164 f"Cannot validate search engine '{engine}': engine configuration unavailable"
165 )
166 return engine
169def _validate_strategy(strategy: Optional[str]) -> Optional[str]:
170 """Validate strategy name against available strategies."""
171 if strategy is None: 171 ↛ 172line 171 didn't jump to line 172 because the condition on line 171 was never true
172 return None
173 strategy = strategy.strip()
174 if not strategy:
175 return None
176 available = get_available_strategies()
177 available_names = [s["name"] for s in available]
178 if strategy not in available_names:
179 raise ValidationError(
180 f"Unknown strategy '{strategy}'. Available: {', '.join(available_names)}"
181 )
182 return strategy
185def _build_settings_overrides(
186 search_engine: Optional[str] = None,
187 strategy: Optional[str] = None,
188 iterations: Optional[int] = None,
189 questions_per_iteration: Optional[int] = None,
190 temperature: Optional[float] = None,
191) -> Dict[str, Any]:
192 """Build settings overrides dict from tool parameters."""
193 overrides: dict[str, Any] = {}
194 if search_engine is not None:
195 search_engine = _validate_search_engine(search_engine)
196 if search_engine:
197 overrides["search.tool"] = search_engine
198 if strategy is not None:
199 strategy = _validate_strategy(strategy)
200 if strategy:
201 overrides["search.search_strategy"] = strategy
202 if iterations is not None:
203 overrides["search.iterations"] = iterations
204 if questions_per_iteration is not None:
205 overrides["search.questions_per_iteration"] = questions_per_iteration
206 if temperature is not None:
207 overrides["llm.temperature"] = temperature
208 return overrides
211# =============================================================================
212# Research Tools
213# =============================================================================
216@mcp.tool()
217def quick_research(
218 query: str,
219 search_engine: Optional[str] = None,
220 strategy: Optional[str] = None,
221 iterations: Optional[int] = None,
222 questions_per_iteration: Optional[int] = None,
223) -> Dict[str, Any]:
224 """
225 Perform quick research on a topic.
227 This tool performs a fast research summary on the given query. It searches
228 the web, analyzes sources, and generates a concise summary with findings.
230 IMPORTANT: This is a synchronous operation that typically takes 1-5 minutes
231 to complete depending on the complexity and configuration.
233 Args:
234 query: The research question or topic to investigate.
235 search_engine: Search engine to use (e.g., "wikipedia", "arxiv", "searxng").
236 Use list_search_engines() to see available options.
237 strategy: Research strategy to use (e.g., "source-based", "rapid", "iterative").
238 Use list_strategies() to see available options.
239 iterations: Number of search iterations (1-10). More iterations = deeper research.
240 questions_per_iteration: Questions to generate per iteration (1-5).
242 Returns:
243 Dictionary containing:
244 - status: "success" or "error"
245 - summary: The research summary text
246 - findings: List of detailed findings from each search
247 - sources: List of source URLs discovered
248 - iterations: Number of iterations performed
249 - error: Error message (only if status is "error")
250 - error_type: Error classification (only if status is "error")
251 """
252 try:
253 # Validate parameters
254 query = _validate_query(query)
255 iterations = _validate_iterations(iterations, max_val=10)
256 questions_per_iteration = _validate_questions_per_iteration(
257 questions_per_iteration
258 )
260 logger.info(f"Starting quick research for query: {query[:100]}...")
262 overrides = _build_settings_overrides(
263 search_engine=search_engine,
264 strategy=strategy,
265 iterations=iterations,
266 questions_per_iteration=questions_per_iteration,
267 )
269 settings = (
270 create_settings_snapshot(overrides=overrides)
271 if overrides
272 else create_settings_snapshot()
273 )
275 result = ldr_quick_summary(query, settings_snapshot=settings)
277 return {
278 "status": "success",
279 "summary": result.get("summary", ""),
280 "findings": result.get("findings", []),
281 "sources": result.get("sources", []),
282 "iterations": result.get("iterations", 0),
283 "formatted_findings": result.get("formatted_findings", ""),
284 }
286 except ValidationError as e:
287 logger.warning("Validation failed for quick research")
288 return {
289 "status": "error",
290 "error": str(e),
291 "error_type": "validation_error",
292 }
293 except Exception as e:
294 logger.exception(
295 f"Quick research failed for query: {query[:100] if query else 'empty'}"
296 )
297 error_type = _classify_error(str(e))
298 return {
299 "status": "error",
300 "error": f"Quick research failed ({error_type}). Check server logs for details.",
301 "error_type": error_type,
302 }
305@mcp.tool()
306def detailed_research(
307 query: str,
308 search_engine: Optional[str] = None,
309 strategy: Optional[str] = None,
310 iterations: Optional[int] = None,
311 questions_per_iteration: Optional[int] = None,
312) -> Dict[str, Any]:
313 """
314 Perform detailed research with comprehensive analysis.
316 This tool performs a thorough research analysis on the given query, returning
317 structured data with detailed findings, sources, and metadata.
319 IMPORTANT: This is a synchronous operation that typically takes 5-15 minutes
320 to complete depending on the complexity and configuration.
322 Args:
323 query: The research question or topic to investigate.
324 search_engine: Search engine to use (e.g., "wikipedia", "arxiv", "searxng").
325 strategy: Research strategy to use (e.g., "source-based", "iterative", "evidence").
326 iterations: Number of search iterations (1-10). More iterations = deeper research.
327 questions_per_iteration: Questions to generate per iteration (1-5).
329 Returns:
330 Dictionary containing:
331 - status: "success" or "error"
332 - query: The original query
333 - research_id: Unique identifier for this research
334 - summary: The research summary text
335 - findings: List of detailed findings
336 - sources: List of source URLs
337 - iterations: Number of iterations performed
338 - metadata: Additional metadata (timestamp, search_tool, strategy)
339 - error/error_type: Error info (only if status is "error")
340 """
341 try:
342 # Validate parameters
343 query = _validate_query(query)
344 iterations = _validate_iterations(iterations, max_val=20)
345 questions_per_iteration = _validate_questions_per_iteration(
346 questions_per_iteration
347 )
349 logger.info(f"Starting detailed research for query: {query[:100]}...")
351 overrides = _build_settings_overrides(
352 search_engine=search_engine,
353 strategy=strategy,
354 iterations=iterations,
355 questions_per_iteration=questions_per_iteration,
356 )
358 settings = (
359 create_settings_snapshot(overrides=overrides)
360 if overrides
361 else create_settings_snapshot()
362 )
364 result = ldr_detailed_research(query, settings_snapshot=settings)
366 return {
367 "status": "success",
368 "query": result.get("query", query),
369 "research_id": result.get("research_id", ""),
370 "summary": result.get("summary", ""),
371 "findings": result.get("findings", []),
372 "sources": result.get("sources", []),
373 "iterations": result.get("iterations", 0),
374 "formatted_findings": result.get("formatted_findings", ""),
375 "metadata": result.get("metadata", {}),
376 }
378 except ValidationError as e:
379 logger.warning("Validation failed for detailed research")
380 return {
381 "status": "error",
382 "error": str(e),
383 "error_type": "validation_error",
384 }
385 except Exception as e:
386 logger.exception(
387 f"Detailed research failed for query: {query[:100] if query else 'empty'}"
388 )
389 error_type = _classify_error(str(e))
390 return {
391 "status": "error",
392 "error": f"Detailed research failed ({error_type}). Check server logs for details.",
393 "error_type": error_type,
394 }
397@mcp.tool()
398def generate_report(
399 query: str,
400 search_engine: Optional[str] = None,
401 searches_per_section: int = 2,
402) -> Dict[str, Any]:
403 """
404 Generate a comprehensive markdown research report.
406 This tool generates a full structured research report with sections,
407 citations, and comprehensive analysis. The output is formatted as markdown.
409 IMPORTANT: This is a synchronous operation that typically takes 10-30 minutes
410 to complete due to the comprehensive nature of the report.
412 Args:
413 query: The research question or topic for the report.
414 search_engine: Search engine to use (e.g., "wikipedia", "arxiv", "searxng").
415 searches_per_section: Number of searches per report section (1-10). Default is 2.
417 Returns:
418 Dictionary containing:
419 - status: "success" or "error"
420 - content: The full report content in markdown format
421 - metadata: Report metadata (timestamp, query)
422 - error/error_type: Error info (only if status is "error")
423 """
424 try:
425 # Validate parameters
426 query = _validate_query(query)
427 if (
428 not isinstance(searches_per_section, int)
429 or searches_per_section < 1
430 ):
431 raise ValidationError( # noqa: TRY301
432 "Searches per section must be a positive integer"
433 )
434 if searches_per_section > 10:
435 raise ValidationError("Searches per section cannot exceed 10") # noqa: TRY301
437 logger.info(f"Starting report generation for query: {query[:100]}...")
439 overrides = {}
440 if search_engine:
441 search_engine = _validate_search_engine(search_engine)
442 if search_engine: 442 ↛ 445line 442 didn't jump to line 445 because the condition on line 442 was always true
443 overrides["search.tool"] = search_engine
445 settings = (
446 create_settings_snapshot(overrides=overrides)
447 if overrides
448 else create_settings_snapshot()
449 )
451 result = ldr_generate_report(
452 query,
453 settings_snapshot=settings,
454 searches_per_section=searches_per_section,
455 )
457 return {
458 "status": "success",
459 "content": result.get("content", ""),
460 "metadata": result.get("metadata", {}),
461 }
463 except ValidationError as e:
464 logger.warning("Validation failed for report generation")
465 return {
466 "status": "error",
467 "error": str(e),
468 "error_type": "validation_error",
469 }
470 except Exception as e:
471 logger.exception(
472 f"Report generation failed for query: {query[:100] if query else 'empty'}"
473 )
474 error_type = _classify_error(str(e))
475 return {
476 "status": "error",
477 "error": f"Report generation failed ({error_type}). Check server logs for details.",
478 "error_type": error_type,
479 }
482@mcp.tool()
483def analyze_documents(
484 query: str,
485 collection_name: str,
486 max_results: int = 10,
487) -> Dict[str, Any]:
488 """
489 Search and analyze documents in a local collection.
491 This tool performs RAG (Retrieval Augmented Generation) search on a
492 local document collection and generates a summary of relevant findings.
494 Args:
495 query: The search query for the documents.
496 collection_name: Name of the local document collection to search.
497 max_results: Maximum number of documents to retrieve (1-100). Default is 10.
499 Returns:
500 Dictionary containing:
501 - status: "success" or "error"
502 - summary: Summary of findings from the documents
503 - documents: List of matching documents with content and metadata
504 - collection: Name of the collection searched
505 - document_count: Number of documents found
506 - error/error_type: Error info (only if status is "error")
507 """
508 try:
509 # Validate parameters
510 query = _validate_query(query)
511 if not collection_name or not collection_name.strip():
512 raise ValidationError("Collection name cannot be empty") # noqa: TRY301
513 collection_name = collection_name.strip()
514 if not _COLLECTION_NAME_RE.match(collection_name): 514 ↛ 515line 514 didn't jump to line 515 because the condition on line 514 was never true
515 raise ValidationError( # noqa: TRY301
516 "Collection name may only contain letters, digits, spaces, hyphens, and underscores (max 100 chars)"
517 )
518 max_results = _validate_max_results(max_results)
520 logger.info(
521 f"Analyzing documents in '{collection_name}' for query: {query[:100]}..."
522 )
524 # Build a settings snapshot the same way the other MCP tools do.
525 # Without this, analyze_documents falls back to JSON defaults +
526 # LDR_* env vars and silently ignores user-configured providers,
527 # API keys, and embedding model. Mirrors quick_research (line 278).
528 settings = create_settings_snapshot()
530 result = ldr_analyze_documents(
531 query=query,
532 collection_name=collection_name,
533 max_results=max_results,
534 settings_snapshot=settings,
535 )
537 return {
538 "status": "success",
539 "summary": result.get("summary", ""),
540 "documents": result.get("documents", []),
541 "collection": result.get("collection", collection_name),
542 "document_count": result.get("document_count", 0),
543 }
545 except ValidationError as e:
546 logger.warning("Validation failed for document analysis")
547 return {
548 "status": "error",
549 "error": str(e),
550 "error_type": "validation_error",
551 }
552 except Exception as e:
553 logger.exception(
554 f"Document analysis failed for collection: {collection_name if collection_name else 'empty'}"
555 )
556 error_type = _classify_error(str(e))
557 return {
558 "status": "error",
559 "error": f"Document analysis failed ({error_type}). Check server logs for details.",
560 "error_type": error_type,
561 }
564@mcp.tool()
565def search(
566 query: str,
567 engine: str,
568 max_results: int = 10,
569) -> Dict[str, Any]:
570 """
571 Search using a specific engine and return raw results without LLM processing.
573 This tool performs a direct search query against the specified engine and
574 returns raw results (title, link, snippet). No LLM is involved, making it
575 fast and free of LLM costs.
577 IMPORTANT: This is a fast operation, typically completing in 5-30 seconds.
579 Args:
580 query: The search query string.
581 engine: Search engine to use (e.g., "arxiv", "wikipedia", "searxng", "brave").
582 This is required — use list_search_engines() to see available options.
583 max_results: Maximum number of results to return (1-100). Default is 10.
585 Returns:
586 Dictionary containing:
587 - status: "success" or "error"
588 - query: The original query
589 - engine: The engine used
590 - result_count: Number of results returned
591 - results: List of results, each with title, link, and snippet
592 - error/error_type: Error info (only if status is "error")
593 """
594 try:
595 # Validate parameters
596 query = _validate_query(query)
597 max_results = _validate_max_results(max_results)
599 # Validate engine is non-empty (required parameter)
600 if not engine or not engine.strip(): 600 ↛ 601line 600 didn't jump to line 601 because the condition on line 600 was never true
601 raise ValidationError( # noqa: TRY301
602 "Engine name cannot be empty. Use list_search_engines() to see available options."
603 )
604 engine = engine.strip()
606 # Create settings snapshot (reused for all steps)
607 settings = create_settings_snapshot()
609 # Validate engine name against available engines
610 from local_deep_research.web_search_engines.search_engines_config import (
611 search_config,
612 )
614 engines_config = search_config(settings_snapshot=settings)
615 if engine not in engines_config:
616 available_names = sorted(engines_config.keys())
617 raise ValidationError( # noqa: TRY301
618 f"Unknown search engine '{engine}'. Available: {', '.join(available_names)}"
619 )
621 # Check API key requirement
622 engine_config = engines_config[engine]
623 if engine_config.get("requires_api_key", False):
624 api_key_setting = settings.get(
625 f"search.engine.web.{engine}.api_key"
626 )
627 api_key = None
628 if api_key_setting: 628 ↛ 629line 628 didn't jump to line 629 because the condition on line 628 was never true
629 api_key = (
630 api_key_setting.get("value")
631 if isinstance(api_key_setting, dict)
632 else api_key_setting
633 )
634 if not api_key: 634 ↛ 641line 634 didn't jump to line 641 because the condition on line 634 was always true
635 raise ValidationError( # noqa: TRY301
636 f"Engine '{engine}' requires an API key. "
637 f"Set the LDR_SEARCH_ENGINE_WEB_{engine.upper()}_API_KEY environment variable "
638 f"or configure it in the UI at search.engine.web.{engine}.api_key"
639 )
641 logger.info(
642 f"Starting search on '{engine}' for query: {query[:100]}..."
643 )
645 # Set thread-local settings context so that engine constructors
646 # which internally call get_llm() or get_setting_from_snapshot()
647 # (e.g., arxiv's JournalReputationFilter) can resolve settings.
648 from local_deep_research.config.thread_settings import (
649 clear_settings_context,
650 set_settings_context,
651 )
652 from local_deep_research.settings.manager import SnapshotSettingsContext
654 set_settings_context(SnapshotSettingsContext(settings))
655 try:
656 return _execute_search(query, engine, max_results, settings)
657 finally:
658 clear_settings_context()
660 except ValidationError as e:
661 logger.warning("Validation failed for search")
662 return {
663 "status": "error",
664 "error": str(e),
665 "error_type": "validation_error",
666 }
667 except Exception as e:
668 logger.exception(
669 f"Search failed for query: {query[:100] if query else 'empty'}"
670 )
671 error_type = _classify_error(str(e))
672 return {
673 "status": "error",
674 "error": f"Search failed ({error_type}). Check server logs for details.",
675 "error_type": error_type,
676 }
679def _egress_audit_net(settings: Dict[str, Any]):
680 """Best-effort context manager that arms the egress audit-hook net for
681 a direct MCP search.
683 Direct MCP searches call ``engine.run()`` without going through
684 ``AdvancedSearchSystem`` (which arms the net itself), so under
685 PRIVATE_ONLY/STRICT the socket-level backstop would otherwise stay
686 inactive for this path. Returns a nullcontext when the policy cannot
687 be built or a context is already armed — the factory PEP remains the
688 primary enforcement, and an unevaluable policy must not break MCP.
689 """
690 from contextlib import nullcontext
692 try:
693 from local_deep_research.security.egress.audit_hook import (
694 active_egress_context,
695 get_active_context,
696 )
697 from local_deep_research.security.egress.policy import (
698 PolicyDeniedError,
699 context_from_snapshot,
700 )
701 except Exception:
702 return nullcontext()
704 if not settings or get_active_context() is not None:
705 return nullcontext()
706 try:
707 primary = unwrap_setting(
708 settings.get("search.tool", DEFAULT_SEARCH_TOOL)
709 )
710 ctx = context_from_snapshot(
711 settings,
712 primary or DEFAULT_SEARCH_TOOL,
713 username=settings.get("_username"),
714 )
715 except (PolicyDeniedError, ValueError):
716 logger.bind(policy_audit=True).debug(
717 "egress audit net not armed for MCP search: policy unevaluable",
718 exc_info=True,
719 )
720 return nullcontext()
721 except Exception:
722 return nullcontext()
723 return active_egress_context(ctx)
726def _execute_search(
727 query: str, engine: str, max_results: int, settings: Dict[str, Any]
728) -> Dict[str, Any]:
729 """Execute the search after settings context is established."""
730 from local_deep_research.web_search_engines.search_engine_factory import (
731 create_search_engine,
732 )
734 search_engine = create_search_engine(
735 engine_name=engine,
736 llm=None,
737 settings_snapshot=settings,
738 programmatic_mode=True,
739 max_results=max_results,
740 search_snippets_only=True,
741 )
743 if search_engine is None:
744 return {
745 "status": "error",
746 "error": f"Failed to create search engine '{engine}'. "
747 f"This engine may require an LLM or have other prerequisites. "
748 f"Check server logs for details.",
749 "error_type": "configuration_error",
750 }
752 try:
753 # Execute search with the egress audit-hook net armed (no-op
754 # under scopes that don't arm it or when policy is unavailable).
755 with _egress_audit_net(settings):
756 results = search_engine.run(query)
758 # Normalize results: ensure consistent 'snippet' key
759 for result in results:
760 if "snippet" not in result and "body" in result:
761 result["snippet"] = result["body"]
763 return {
764 "status": "success",
765 "query": query,
766 "engine": engine,
767 "result_count": len(results),
768 "results": results,
769 }
770 finally:
771 from local_deep_research.utilities.resource_utils import safe_close
773 safe_close(search_engine, "MCP search engine")
776# =============================================================================
777# Discovery Tools
778# =============================================================================
781@mcp.tool()
782def list_search_engines() -> Dict[str, Any]:
783 """
784 List available search engines.
786 Returns a list of search engines that can be used with the research tools.
787 Each engine has different strengths - some are better for academic research,
788 others for current events, etc.
790 Returns:
791 Dictionary containing:
792 - status: "success" or "error"
793 - engines: List of available search engine configurations
794 - error/error_type: Error info (only if status is "error")
795 """
796 try:
797 from local_deep_research.api.settings_utils import (
798 create_settings_snapshot,
799 )
800 from local_deep_research.web_search_engines.search_engines_config import (
801 search_config,
802 )
804 settings = create_settings_snapshot()
805 engines_config = search_config(settings_snapshot=settings)
807 engines = []
808 for name, config in engines_config.items():
809 engine_info = {
810 "name": name,
811 "description": config.get("description", ""),
812 "strengths": config.get("strengths", []),
813 "weaknesses": config.get("weaknesses", []),
814 "requires_api_key": config.get("requires_api_key", False),
815 "is_local": config.get("is_local", False),
816 }
817 engines.append(engine_info)
819 return {
820 "status": "success",
821 "engines": sorted(engines, key=lambda x: x["name"]),
822 }
824 except Exception as e:
825 logger.exception("Failed to list search engines")
826 error_type = _classify_error(str(e))
827 return {
828 "status": "error",
829 "error": f"Failed to list search engines ({error_type}). Check server logs for details.",
830 "error_type": error_type,
831 }
834@mcp.tool()
835def list_strategies() -> Dict[str, Any]:
836 """
837 List available research strategies.
839 Returns a list of research strategies that can be used with the research tools.
840 Each strategy has different characteristics suited for different types of queries.
842 Returns:
843 Dictionary containing:
844 - status: "success" or "error"
845 - strategies: List of available strategies with names and descriptions
846 - error/error_type: Error info (only if status is "error")
847 """
848 try:
849 return {
850 "status": "success",
851 "strategies": get_available_strategies(),
852 }
854 except Exception as e:
855 logger.exception("Failed to list strategies")
856 error_type = _classify_error(str(e))
857 return {
858 "status": "error",
859 "error": f"Failed to list strategies ({error_type}). Check server logs for details.",
860 "error_type": error_type,
861 }
864@mcp.tool()
865def get_configuration() -> Dict[str, Any]:
866 """
867 Get current server configuration.
869 Returns the current configuration settings being used by the MCP server,
870 including LLM provider, default search engine, and other settings.
872 Returns:
873 Dictionary containing:
874 - status: "success" or "error"
875 - config: Current configuration settings
876 - error/error_type: Error info (only if status is "error")
877 """
878 try:
879 from local_deep_research.api.settings_utils import (
880 create_settings_snapshot,
881 extract_setting_value,
882 )
884 settings = create_settings_snapshot()
886 config = {
887 "llm": {
888 "provider": extract_setting_value(
889 settings, "llm.provider", "unknown"
890 ),
891 "model": extract_setting_value(
892 settings, "llm.model", "unknown"
893 ),
894 "temperature": extract_setting_value(
895 settings, "llm.temperature", 0.7
896 ),
897 },
898 "search": {
899 "default_engine": extract_setting_value(
900 settings, "search.tool", DEFAULT_SEARCH_TOOL
901 ),
902 "default_strategy": extract_setting_value(
903 settings, "search.search_strategy", "source-based"
904 ),
905 "iterations": extract_setting_value(
906 settings, "search.iterations", 2
907 ),
908 "questions_per_iteration": extract_setting_value(
909 settings, "search.questions_per_iteration", 3
910 ),
911 "max_results": extract_setting_value(
912 settings, "search.max_results", 10
913 ),
914 },
915 }
917 return {
918 "status": "success",
919 "config": config,
920 }
922 except Exception as e:
923 logger.exception("Failed to get configuration")
924 error_type = _classify_error(str(e))
925 return {
926 "status": "error",
927 "error": f"Failed to get configuration ({error_type}). Check server logs for details.",
928 "error_type": error_type,
929 }
932# =============================================================================
933# Server Entry Point
934# =============================================================================
937def run_server():
938 """Run the MCP server using STDIO transport."""
939 # MCP uses stdout for JSON-RPC, so redirect all logging to stderr.
940 # This runs in a separate subprocess (ldr-mcp) — logger.remove() only
941 # affects this MCP process, not the main LDR application.
942 logger.remove()
943 # diagnose=False: loguru's default is True, which renders repr() of every
944 # local in every traceback frame on exception. The many logger.exception()
945 # call sites in this file run with frame locals that hold credentials
946 # (api_key, Authorization headers, search-engine secrets), so leaving the
947 # default on would write them to the MCP client's stderr log on any
948 # failure. Companion to #4185 / config_logger's LDR_LOGURU_DIAGNOSE gate;
949 # the MCP subprocess has no debug mode, so the gate is unconditionally off.
950 logger.add(
951 sys.stderr,
952 level="INFO",
953 format="{time:YYYY-MM-DD HH:mm:ss} | {level: <8} | {name}:{function}:{line} | {message}",
954 diagnose=False,
955 )
956 logger.info("Starting Local Deep Research MCP server...")
958 # Phase-1 of the RAG plaintext-at-rest migration must run here too, not only
959 # in the ldr-web entrypoint: an MCP-only deployment (Claude Desktop / Code /
960 # OpenClaw) that never launches ldr-web would otherwise never purge a
961 # pre-existing legacy plaintext <hash>.pkl docstore, leaving chunk text on
962 # disk indefinitely. The sweep is filesystem-only (no DB/login needed),
963 # idempotent (a clean tree no-ops), and logs its own errors — wrapped so a
964 # cleanup issue never blocks the server. Mirrors web/app.py's main().
965 try:
966 from ..vector_stores.legacy_cleanup import migrate_legacy_docstores
968 migrate_legacy_docstores()
969 except Exception:
970 logger.exception("Legacy RAG docstore migration failed at MCP startup")
972 mcp.run(transport="stdio")
975if __name__ == "__main__": 975 ↛ 976line 975 didn't jump to line 976 because the condition on line 975 was never true
976 run_server()