search_and_merge
Execute multi-source literature searches and merge results using predefined query plans to consolidate biomedical research findings.
Instructions
Run multi-source search and merge using the existing query plan.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| task_id | Yes | ||
| per_query | No |
Output Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
No arguments | |||
Implementation Reference
- The search_and_merge tool handler function. Takes task_id and per_query parameters, runs the backend CLI command 'step2-search-and-merge', and returns search results including paths to master_results.jsonl, screening_table.csv, and source statistics.
@mcp.tool def search_and_merge(task_id: str, per_query: int = 30) -> dict[str, Any]: """Run multi-source search and merge using the existing query plan.""" root = task_root(task_id) plan_path = root / "query_plan.json" out_dir = root / "search" result = ensure_success( run_backend( [ "step2-search-and-merge", "--plan", str(plan_path), "--out", str(out_dir), "--per-query", str(per_query), ] ) ) stats_path = out_dir / "source_stats.json" stats = json.loads(stats_path.read_text()) if stats_path.exists() else {} return { **result, "task_id": task_id, "search_dir": str(out_dir), "master_results_path": str(out_dir / "master_results.jsonl"), "screening_table_csv": str(out_dir / "screening_table.csv"), "screening_table_xlsx": str(out_dir / "screening_table.xlsx"), "oa_candidates_path": str(out_dir / "oa_candidates.jsonl"), "source_stats": stats, } - src/literature_agent_mcp/server.py:93-93 (registration)FastMCP tool registration decorator. The @mcp.tool decorator registers search_and_merge as an MCP tool, making it available to the MCP server.
@mcp.tool - Helper functions that support search_and_merge execution. run_backend() executes CLI commands in the literature-agent environment, ensure_success() validates command exit codes, and task_root() resolves task directories.
def run_backend(args: list[str]) -> dict[str, Any]: root = literature_agent_root() if not root.exists(): raise FileNotFoundError(f"literature-agent root not found: {root}") cmd = [backend_python(root), "-m", "litagent.cli", *args] proc = subprocess.run( cmd, cwd=root, capture_output=True, text=True, check=False, ) return { "cwd": str(root), "command": cmd, "returncode": proc.returncode, "stdout": proc.stdout, "stderr": proc.stderr, } def task_root(task_id: str) -> Path: return literature_agent_root() / "work" / task_id def ensure_success(result: dict[str, Any]) -> dict[str, Any]: if result["returncode"] != 0: raise RuntimeError( f"backend command failed ({result['returncode']}):\nSTDOUT:\n{result['stdout']}\nSTDERR:\n{result['stderr']}" ) return result