Coverage for src/local_deep_research/mcp/server.py: 94%
285 statements
« prev ^ index » next coverage.py v7.14.1, created at 2026-06-03 23:15 +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)
50# Create FastMCP server instance
51mcp = FastMCP(
52 "local-deep-research",
53 instructions="AI-powered deep research assistant with iterative analysis using LLMs and web searches",
54)
57def _classify_error(error_msg: str) -> str:
58 """Classify error for client handling."""
59 error_lower = error_msg.lower()
60 if "503" in error_msg or "unavailable" in error_lower:
61 return "service_unavailable"
62 if "404" in error_msg or "not found" in error_lower:
63 return "model_not_found"
64 if (
65 "api key" in error_lower
66 or "authentication" in error_lower
67 or "unauthorized" in error_lower
68 or "401" in error_msg
69 ):
70 return "auth_error"
71 if "timeout" in error_lower or "timed out" in error_lower:
72 return "timeout"
73 if "rate limit" in error_lower or "429" in error_msg:
74 return "rate_limit"
75 if "connection" in error_lower:
76 return "connection_error"
77 if "validation" in error_lower or "invalid" in error_lower:
78 return "validation_error"
79 return "unknown"
82class ValidationError(Exception):
83 """Raised when parameter validation fails."""
85 pass
88_COLLECTION_NAME_RE = re.compile(r"^[A-Za-z0-9 _-]{1,100}$")
91def _validate_query(query: str) -> str:
92 """Validate and sanitize query parameter."""
93 if not query or not query.strip():
94 raise ValidationError("Query cannot be empty")
95 query = query.strip()
96 if len(query) > 10000:
97 raise ValidationError(
98 "Query exceeds maximum length of 10000 characters"
99 )
100 return query
103def _validate_iterations(
104 iterations: Optional[int], max_val: int = 20
105) -> Optional[int]:
106 """Validate iterations parameter."""
107 if iterations is None:
108 return None
109 if not isinstance(iterations, int) or iterations < 1:
110 raise ValidationError("Iterations must be a positive integer")
111 if iterations > max_val:
112 raise ValidationError(f"Iterations cannot exceed {max_val}")
113 return iterations
116def _validate_questions_per_iteration(qpi: Optional[int]) -> Optional[int]:
117 """Validate questions_per_iteration parameter."""
118 if qpi is None:
119 return None
120 if not isinstance(qpi, int) or qpi < 1:
121 raise ValidationError(
122 "Questions per iteration must be a positive integer"
123 )
124 if qpi > 10:
125 raise ValidationError("Questions per iteration cannot exceed 10")
126 return qpi
129def _validate_max_results(max_results: int) -> int:
130 """Validate max_results parameter."""
131 if not isinstance(max_results, int) or max_results < 1:
132 raise ValidationError("Max results must be a positive integer")
133 if max_results > 100:
134 raise ValidationError("Max results cannot exceed 100")
135 return max_results
138def _validate_temperature(temperature: Optional[float]) -> Optional[float]:
139 """Validate temperature parameter."""
140 if temperature is None:
141 return None
142 if not isinstance(temperature, (int, float)):
143 raise ValidationError("Temperature must be a number")
144 if temperature < 0.0 or temperature > 2.0:
145 raise ValidationError("Temperature must be between 0.0 and 2.0")
146 return float(temperature)
149def _validate_search_engine(engine: Optional[str]) -> Optional[str]:
150 """Validate search engine name against available engines."""
151 if engine is None:
152 return None
153 engine = engine.strip()
154 if not engine:
155 return None
156 try:
157 from local_deep_research.web_search_engines.search_engines_config import (
158 search_config,
159 )
161 settings = create_settings_snapshot()
162 available = search_config(settings_snapshot=settings)
163 if engine not in available:
164 available_names = sorted(available.keys())
165 raise ValidationError( # noqa: TRY301
166 f"Unknown search engine '{engine}'. Available: {', '.join(available_names)}"
167 )
168 except ValidationError:
169 raise
170 except Exception:
171 logger.exception("Could not load engine config to validate engine")
172 raise ValidationError(
173 f"Cannot validate search engine '{engine}': engine configuration unavailable"
174 )
175 return engine
178def _validate_strategy(strategy: Optional[str]) -> Optional[str]:
179 """Validate strategy name against available strategies."""
180 if strategy is None: 180 ↛ 181line 180 didn't jump to line 181 because the condition on line 180 was never true
181 return None
182 strategy = strategy.strip()
183 if not strategy:
184 return None
185 available = get_available_strategies(show_all=True)
186 available_names = [s["name"] for s in available]
187 if strategy not in available_names:
188 raise ValidationError(
189 f"Unknown strategy '{strategy}'. Available: {', '.join(available_names)}"
190 )
191 return strategy
194def _build_settings_overrides(
195 search_engine: Optional[str] = None,
196 strategy: Optional[str] = None,
197 iterations: Optional[int] = None,
198 questions_per_iteration: Optional[int] = None,
199 temperature: Optional[float] = None,
200) -> Dict[str, Any]:
201 """Build settings overrides dict from tool parameters."""
202 overrides: dict[str, Any] = {}
203 if search_engine is not None:
204 search_engine = _validate_search_engine(search_engine)
205 if search_engine:
206 overrides["search.tool"] = search_engine
207 if strategy is not None:
208 strategy = _validate_strategy(strategy)
209 if strategy:
210 overrides["search.search_strategy"] = strategy
211 if iterations is not None:
212 overrides["search.iterations"] = iterations
213 if questions_per_iteration is not None:
214 overrides["search.questions_per_iteration"] = questions_per_iteration
215 if temperature is not None:
216 overrides["llm.temperature"] = temperature
217 return overrides
220# =============================================================================
221# Research Tools
222# =============================================================================
225@mcp.tool()
226def quick_research(
227 query: str,
228 search_engine: Optional[str] = None,
229 strategy: Optional[str] = None,
230 iterations: Optional[int] = None,
231 questions_per_iteration: Optional[int] = None,
232) -> Dict[str, Any]:
233 """
234 Perform quick research on a topic.
236 This tool performs a fast research summary on the given query. It searches
237 the web, analyzes sources, and generates a concise summary with findings.
239 IMPORTANT: This is a synchronous operation that typically takes 1-5 minutes
240 to complete depending on the complexity and configuration.
242 Args:
243 query: The research question or topic to investigate.
244 search_engine: Search engine to use (e.g., "wikipedia", "arxiv", "searxng", "auto").
245 Use list_search_engines() to see available options.
246 strategy: Research strategy to use (e.g., "source-based", "rapid", "iterative").
247 Use list_strategies() to see available options.
248 iterations: Number of search iterations (1-10). More iterations = deeper research.
249 questions_per_iteration: Questions to generate per iteration (1-5).
251 Returns:
252 Dictionary containing:
253 - status: "success" or "error"
254 - summary: The research summary text
255 - findings: List of detailed findings from each search
256 - sources: List of source URLs discovered
257 - iterations: Number of iterations performed
258 - error: Error message (only if status is "error")
259 - error_type: Error classification (only if status is "error")
260 """
261 try:
262 # Validate parameters
263 query = _validate_query(query)
264 iterations = _validate_iterations(iterations, max_val=10)
265 questions_per_iteration = _validate_questions_per_iteration(
266 questions_per_iteration
267 )
269 logger.info(f"Starting quick research for query: {query[:100]}...")
271 overrides = _build_settings_overrides(
272 search_engine=search_engine,
273 strategy=strategy,
274 iterations=iterations,
275 questions_per_iteration=questions_per_iteration,
276 )
278 settings = (
279 create_settings_snapshot(overrides=overrides)
280 if overrides
281 else create_settings_snapshot()
282 )
284 result = ldr_quick_summary(query, settings_snapshot=settings)
286 return {
287 "status": "success",
288 "summary": result.get("summary", ""),
289 "findings": result.get("findings", []),
290 "sources": result.get("sources", []),
291 "iterations": result.get("iterations", 0),
292 "formatted_findings": result.get("formatted_findings", ""),
293 }
295 except ValidationError as e:
296 logger.warning("Validation failed for quick research")
297 return {
298 "status": "error",
299 "error": str(e),
300 "error_type": "validation_error",
301 }
302 except Exception as e:
303 logger.exception(
304 f"Quick research failed for query: {query[:100] if query else 'empty'}"
305 )
306 error_type = _classify_error(str(e))
307 return {
308 "status": "error",
309 "error": f"Quick research failed ({error_type}). Check server logs for details.",
310 "error_type": error_type,
311 }
314@mcp.tool()
315def detailed_research(
316 query: str,
317 search_engine: Optional[str] = None,
318 strategy: Optional[str] = None,
319 iterations: Optional[int] = None,
320 questions_per_iteration: Optional[int] = None,
321) -> Dict[str, Any]:
322 """
323 Perform detailed research with comprehensive analysis.
325 This tool performs a thorough research analysis on the given query, returning
326 structured data with detailed findings, sources, and metadata.
328 IMPORTANT: This is a synchronous operation that typically takes 5-15 minutes
329 to complete depending on the complexity and configuration.
331 Args:
332 query: The research question or topic to investigate.
333 search_engine: Search engine to use (e.g., "wikipedia", "arxiv", "searxng", "auto").
334 strategy: Research strategy to use (e.g., "source-based", "iterative", "evidence").
335 iterations: Number of search iterations (1-10). More iterations = deeper research.
336 questions_per_iteration: Questions to generate per iteration (1-5).
338 Returns:
339 Dictionary containing:
340 - status: "success" or "error"
341 - query: The original query
342 - research_id: Unique identifier for this research
343 - summary: The research summary text
344 - findings: List of detailed findings
345 - sources: List of source URLs
346 - iterations: Number of iterations performed
347 - metadata: Additional metadata (timestamp, search_tool, strategy)
348 - error/error_type: Error info (only if status is "error")
349 """
350 try:
351 # Validate parameters
352 query = _validate_query(query)
353 iterations = _validate_iterations(iterations, max_val=20)
354 questions_per_iteration = _validate_questions_per_iteration(
355 questions_per_iteration
356 )
358 logger.info(f"Starting detailed research for query: {query[:100]}...")
360 overrides = _build_settings_overrides(
361 search_engine=search_engine,
362 strategy=strategy,
363 iterations=iterations,
364 questions_per_iteration=questions_per_iteration,
365 )
367 settings = (
368 create_settings_snapshot(overrides=overrides)
369 if overrides
370 else create_settings_snapshot()
371 )
373 result = ldr_detailed_research(query, settings_snapshot=settings)
375 return {
376 "status": "success",
377 "query": result.get("query", query),
378 "research_id": result.get("research_id", ""),
379 "summary": result.get("summary", ""),
380 "findings": result.get("findings", []),
381 "sources": result.get("sources", []),
382 "iterations": result.get("iterations", 0),
383 "formatted_findings": result.get("formatted_findings", ""),
384 "metadata": result.get("metadata", {}),
385 }
387 except ValidationError as e:
388 logger.warning("Validation failed for detailed research")
389 return {
390 "status": "error",
391 "error": str(e),
392 "error_type": "validation_error",
393 }
394 except Exception as e:
395 logger.exception(
396 f"Detailed research failed for query: {query[:100] if query else 'empty'}"
397 )
398 error_type = _classify_error(str(e))
399 return {
400 "status": "error",
401 "error": f"Detailed research failed ({error_type}). Check server logs for details.",
402 "error_type": error_type,
403 }
406@mcp.tool()
407def generate_report(
408 query: str,
409 search_engine: Optional[str] = None,
410 searches_per_section: int = 2,
411) -> Dict[str, Any]:
412 """
413 Generate a comprehensive markdown research report.
415 This tool generates a full structured research report with sections,
416 citations, and comprehensive analysis. The output is formatted as markdown.
418 IMPORTANT: This is a synchronous operation that typically takes 10-30 minutes
419 to complete due to the comprehensive nature of the report.
421 Args:
422 query: The research question or topic for the report.
423 search_engine: Search engine to use (e.g., "wikipedia", "arxiv", "searxng", "auto").
424 searches_per_section: Number of searches per report section (1-10). Default is 2.
426 Returns:
427 Dictionary containing:
428 - status: "success" or "error"
429 - content: The full report content in markdown format
430 - metadata: Report metadata (timestamp, query)
431 - error/error_type: Error info (only if status is "error")
432 """
433 try:
434 # Validate parameters
435 query = _validate_query(query)
436 if (
437 not isinstance(searches_per_section, int)
438 or searches_per_section < 1
439 ):
440 raise ValidationError( # noqa: TRY301
441 "Searches per section must be a positive integer"
442 )
443 if searches_per_section > 10:
444 raise ValidationError("Searches per section cannot exceed 10") # noqa: TRY301
446 logger.info(f"Starting report generation for query: {query[:100]}...")
448 overrides = {}
449 if search_engine:
450 search_engine = _validate_search_engine(search_engine)
451 if search_engine: 451 ↛ 454line 451 didn't jump to line 454 because the condition on line 451 was always true
452 overrides["search.tool"] = search_engine
454 settings = (
455 create_settings_snapshot(overrides=overrides)
456 if overrides
457 else create_settings_snapshot()
458 )
460 result = ldr_generate_report(
461 query,
462 settings_snapshot=settings,
463 searches_per_section=searches_per_section,
464 )
466 return {
467 "status": "success",
468 "content": result.get("content", ""),
469 "metadata": result.get("metadata", {}),
470 }
472 except ValidationError as e:
473 logger.warning("Validation failed for report generation")
474 return {
475 "status": "error",
476 "error": str(e),
477 "error_type": "validation_error",
478 }
479 except Exception as e:
480 logger.exception(
481 f"Report generation failed for query: {query[:100] if query else 'empty'}"
482 )
483 error_type = _classify_error(str(e))
484 return {
485 "status": "error",
486 "error": f"Report generation failed ({error_type}). Check server logs for details.",
487 "error_type": error_type,
488 }
491@mcp.tool()
492def analyze_documents(
493 query: str,
494 collection_name: str,
495 max_results: int = 10,
496) -> Dict[str, Any]:
497 """
498 Search and analyze documents in a local collection.
500 This tool performs RAG (Retrieval Augmented Generation) search on a
501 local document collection and generates a summary of relevant findings.
503 Args:
504 query: The search query for the documents.
505 collection_name: Name of the local document collection to search.
506 max_results: Maximum number of documents to retrieve (1-100). Default is 10.
508 Returns:
509 Dictionary containing:
510 - status: "success" or "error"
511 - summary: Summary of findings from the documents
512 - documents: List of matching documents with content and metadata
513 - collection: Name of the collection searched
514 - document_count: Number of documents found
515 - error/error_type: Error info (only if status is "error")
516 """
517 try:
518 # Validate parameters
519 query = _validate_query(query)
520 if not collection_name or not collection_name.strip():
521 raise ValidationError("Collection name cannot be empty") # noqa: TRY301
522 collection_name = collection_name.strip()
523 if not _COLLECTION_NAME_RE.match(collection_name): 523 ↛ 524line 523 didn't jump to line 524 because the condition on line 523 was never true
524 raise ValidationError( # noqa: TRY301
525 "Collection name may only contain letters, digits, spaces, hyphens, and underscores (max 100 chars)"
526 )
527 max_results = _validate_max_results(max_results)
529 logger.info(
530 f"Analyzing documents in '{collection_name}' for query: {query[:100]}..."
531 )
533 # Build a settings snapshot the same way the other MCP tools do.
534 # Without this, analyze_documents falls back to JSON defaults +
535 # LDR_* env vars and silently ignores user-configured providers,
536 # API keys, and embedding model. Mirrors quick_research (line 278).
537 settings = create_settings_snapshot()
539 result = ldr_analyze_documents(
540 query=query,
541 collection_name=collection_name,
542 max_results=max_results,
543 settings_snapshot=settings,
544 )
546 return {
547 "status": "success",
548 "summary": result.get("summary", ""),
549 "documents": result.get("documents", []),
550 "collection": result.get("collection", collection_name),
551 "document_count": result.get("document_count", 0),
552 }
554 except ValidationError as e:
555 logger.warning("Validation failed for document analysis")
556 return {
557 "status": "error",
558 "error": str(e),
559 "error_type": "validation_error",
560 }
561 except Exception as e:
562 logger.exception(
563 f"Document analysis failed for collection: {collection_name if collection_name else 'empty'}"
564 )
565 error_type = _classify_error(str(e))
566 return {
567 "status": "error",
568 "error": f"Document analysis failed ({error_type}). Check server logs for details.",
569 "error_type": error_type,
570 }
573@mcp.tool()
574def search(
575 query: str,
576 engine: str,
577 max_results: int = 10,
578) -> Dict[str, Any]:
579 """
580 Search using a specific engine and return raw results without LLM processing.
582 This tool performs a direct search query against the specified engine and
583 returns raw results (title, link, snippet). No LLM is involved, making it
584 fast and free of LLM costs.
586 IMPORTANT: This is a fast operation, typically completing in 5-30 seconds.
588 Args:
589 query: The search query string.
590 engine: Search engine to use (e.g., "arxiv", "wikipedia", "searxng", "brave").
591 This is required — use list_search_engines() to see available options.
592 max_results: Maximum number of results to return (1-100). Default is 10.
594 Returns:
595 Dictionary containing:
596 - status: "success" or "error"
597 - query: The original query
598 - engine: The engine used
599 - result_count: Number of results returned
600 - results: List of results, each with title, link, and snippet
601 - error/error_type: Error info (only if status is "error")
602 """
603 try:
604 # Validate parameters
605 query = _validate_query(query)
606 max_results = _validate_max_results(max_results)
608 # Validate engine is non-empty (required parameter)
609 if not engine or not engine.strip(): 609 ↛ 610line 609 didn't jump to line 610 because the condition on line 609 was never true
610 raise ValidationError( # noqa: TRY301
611 "Engine name cannot be empty. Use list_search_engines() to see available options."
612 )
613 engine = engine.strip()
615 # Create settings snapshot (reused for all steps)
616 settings = create_settings_snapshot()
618 # Validate engine name against available engines
619 from local_deep_research.web_search_engines.search_engines_config import (
620 search_config,
621 )
623 engines_config = search_config(settings_snapshot=settings)
624 if engine not in engines_config:
625 available_names = sorted(engines_config.keys())
626 raise ValidationError( # noqa: TRY301
627 f"Unknown search engine '{engine}'. Available: {', '.join(available_names)}"
628 )
630 # Check API key requirement
631 engine_config = engines_config[engine]
632 if engine_config.get("requires_api_key", False):
633 api_key_setting = settings.get(
634 f"search.engine.web.{engine}.api_key"
635 )
636 api_key = None
637 if api_key_setting: 637 ↛ 638line 637 didn't jump to line 638 because the condition on line 637 was never true
638 api_key = (
639 api_key_setting.get("value")
640 if isinstance(api_key_setting, dict)
641 else api_key_setting
642 )
643 if not api_key: 643 ↛ 650line 643 didn't jump to line 650 because the condition on line 643 was always true
644 raise ValidationError( # noqa: TRY301
645 f"Engine '{engine}' requires an API key. "
646 f"Set the LDR_SEARCH_ENGINE_WEB_{engine.upper()}_API_KEY environment variable "
647 f"or configure it in the UI at search.engine.web.{engine}.api_key"
648 )
650 logger.info(
651 f"Starting search on '{engine}' for query: {query[:100]}..."
652 )
654 # Set thread-local settings context so that engine constructors
655 # which internally call get_llm() or get_setting_from_snapshot()
656 # (e.g., arxiv's JournalReputationFilter) can resolve settings.
657 from local_deep_research.config.thread_settings import (
658 clear_settings_context,
659 set_settings_context,
660 )
661 from local_deep_research.settings.manager import SnapshotSettingsContext
663 set_settings_context(SnapshotSettingsContext(settings))
664 try:
665 return _execute_search(query, engine, max_results, settings)
666 finally:
667 clear_settings_context()
669 except ValidationError as e:
670 logger.warning("Validation failed for search")
671 return {
672 "status": "error",
673 "error": str(e),
674 "error_type": "validation_error",
675 }
676 except Exception as e:
677 logger.exception(
678 f"Search failed for query: {query[:100] if query else 'empty'}"
679 )
680 error_type = _classify_error(str(e))
681 return {
682 "status": "error",
683 "error": f"Search failed ({error_type}). Check server logs for details.",
684 "error_type": error_type,
685 }
688def _execute_search(
689 query: str, engine: str, max_results: int, settings: Dict[str, Any]
690) -> Dict[str, Any]:
691 """Execute the search after settings context is established."""
692 from local_deep_research.web_search_engines.search_engine_factory import (
693 create_search_engine,
694 )
696 search_engine = create_search_engine(
697 engine_name=engine,
698 llm=None,
699 settings_snapshot=settings,
700 programmatic_mode=True,
701 max_results=max_results,
702 search_snippets_only=True,
703 )
705 if search_engine is None:
706 return {
707 "status": "error",
708 "error": f"Failed to create search engine '{engine}'. "
709 f"This engine may require an LLM or have other prerequisites. "
710 f"Check server logs for details.",
711 "error_type": "configuration_error",
712 }
714 try:
715 # Execute search
716 results = search_engine.run(query)
718 # Normalize results: ensure consistent 'snippet' key
719 for result in results:
720 if "snippet" not in result and "body" in result:
721 result["snippet"] = result["body"]
723 return {
724 "status": "success",
725 "query": query,
726 "engine": engine,
727 "result_count": len(results),
728 "results": results,
729 }
730 finally:
731 from local_deep_research.utilities.resource_utils import safe_close
733 safe_close(search_engine, "MCP search engine")
736# =============================================================================
737# Discovery Tools
738# =============================================================================
741@mcp.tool()
742def list_search_engines() -> Dict[str, Any]:
743 """
744 List available search engines.
746 Returns a list of search engines that can be used with the research tools.
747 Each engine has different strengths - some are better for academic research,
748 others for current events, etc.
750 Returns:
751 Dictionary containing:
752 - status: "success" or "error"
753 - engines: List of available search engine configurations
754 - error/error_type: Error info (only if status is "error")
755 """
756 try:
757 from local_deep_research.api.settings_utils import (
758 create_settings_snapshot,
759 )
760 from local_deep_research.web_search_engines.search_engines_config import (
761 search_config,
762 )
764 settings = create_settings_snapshot()
765 engines_config = search_config(settings_snapshot=settings)
767 engines = []
768 for name, config in engines_config.items():
769 engine_info = {
770 "name": name,
771 "description": config.get("description", ""),
772 "strengths": config.get("strengths", []),
773 "weaknesses": config.get("weaknesses", []),
774 "requires_api_key": config.get("requires_api_key", False),
775 "is_local": config.get("is_local", False),
776 }
777 engines.append(engine_info)
779 return {
780 "status": "success",
781 "engines": sorted(engines, key=lambda x: x["name"]),
782 }
784 except Exception as e:
785 logger.exception("Failed to list search engines")
786 error_type = _classify_error(str(e))
787 return {
788 "status": "error",
789 "error": f"Failed to list search engines ({error_type}). Check server logs for details.",
790 "error_type": error_type,
791 }
794@mcp.tool()
795def list_strategies() -> Dict[str, Any]:
796 """
797 List available research strategies.
799 Returns a list of research strategies that can be used with the research tools.
800 Each strategy has different characteristics suited for different types of queries.
802 Returns:
803 Dictionary containing:
804 - status: "success" or "error"
805 - strategies: List of available strategies with names and descriptions
806 - error/error_type: Error info (only if status is "error")
807 """
808 try:
809 return {
810 "status": "success",
811 "strategies": get_available_strategies(show_all=True),
812 }
814 except Exception as e:
815 logger.exception("Failed to list strategies")
816 error_type = _classify_error(str(e))
817 return {
818 "status": "error",
819 "error": f"Failed to list strategies ({error_type}). Check server logs for details.",
820 "error_type": error_type,
821 }
824@mcp.tool()
825def get_configuration() -> Dict[str, Any]:
826 """
827 Get current server configuration.
829 Returns the current configuration settings being used by the MCP server,
830 including LLM provider, default search engine, and other settings.
832 Returns:
833 Dictionary containing:
834 - status: "success" or "error"
835 - config: Current configuration settings
836 - error/error_type: Error info (only if status is "error")
837 """
838 try:
839 from local_deep_research.api.settings_utils import (
840 create_settings_snapshot,
841 extract_setting_value,
842 )
844 settings = create_settings_snapshot()
846 config = {
847 "llm": {
848 "provider": extract_setting_value(
849 settings, "llm.provider", "unknown"
850 ),
851 "model": extract_setting_value(
852 settings, "llm.model", "unknown"
853 ),
854 "temperature": extract_setting_value(
855 settings, "llm.temperature", 0.7
856 ),
857 },
858 "search": {
859 "default_engine": extract_setting_value(
860 settings, "search.tool", "auto"
861 ),
862 "default_strategy": extract_setting_value(
863 settings, "search.search_strategy", "source-based"
864 ),
865 "iterations": extract_setting_value(
866 settings, "search.iterations", 2
867 ),
868 "questions_per_iteration": extract_setting_value(
869 settings, "search.questions_per_iteration", 3
870 ),
871 "max_results": extract_setting_value(
872 settings, "search.max_results", 10
873 ),
874 },
875 }
877 return {
878 "status": "success",
879 "config": config,
880 }
882 except Exception as e:
883 logger.exception("Failed to get configuration")
884 error_type = _classify_error(str(e))
885 return {
886 "status": "error",
887 "error": f"Failed to get configuration ({error_type}). Check server logs for details.",
888 "error_type": error_type,
889 }
892# =============================================================================
893# Server Entry Point
894# =============================================================================
897def run_server():
898 """Run the MCP server using STDIO transport."""
899 # MCP uses stdout for JSON-RPC, so redirect all logging to stderr.
900 # This runs in a separate subprocess (ldr-mcp) — logger.remove() only
901 # affects this MCP process, not the main LDR application.
902 logger.remove()
903 logger.add(
904 sys.stderr,
905 level="INFO",
906 format="{time:YYYY-MM-DD HH:mm:ss} | {level: <8} | {name}:{function}:{line} | {message}",
907 )
908 logger.info("Starting Local Deep Research MCP server...")
909 mcp.run(transport="stdio")
912if __name__ == "__main__": 912 ↛ 913line 912 didn't jump to line 913 because the condition on line 912 was never true
913 run_server()