1use std::path::PathBuf;
11use std::sync::Arc;
12
13use rmcp::handler::server::wrapper::Parameters;
14use rmcp::model::CallToolResult;
15use rmcp::tool;
16use rmcp::tool_handler;
17use rmcp::tool_router;
18use rmcp::{ServerHandler, ServiceExt};
19use tokio::io::AsyncReadExt;
20
21use super::guard;
22use serde_json::{Value, json};
23
24use crate::adapters::arxiv_source::ArxivSource;
25use crate::adapters::europepmc_source::{EuropePmcSource, PreprintSource};
26use crate::adapters::openalex_source::OpenAlexSource;
27use crate::adapters::pdf_source::PdfSource;
28use crate::adapters::semantic_scholar_source::SemanticScholarSource;
29use crate::adapters::sqlite_store::SqliteStore;
30use crate::application::gap_analyzer::{collect_brief, record_gaps};
31use crate::application::ingest_pipeline::IngestPipeline;
32use crate::application::report_generator::{collect_material, save_report};
33use crate::composition::{load_config, open_store};
34use crate::domain::paper::{Paper, Rating, ReadingStatus};
35use crate::domain::research_topic::ResearchTopic;
36use crate::error::ResearchError;
37use crate::ports::index_store::IndexStore;
38
39use super::params::*;
40
41pub struct ResearchContext {
43 pub db_path: PathBuf,
45}
46
47#[derive(Clone)]
50pub struct ResearchServer {
51 ctx: Arc<ResearchContext>,
52}
53
54impl ResearchServer {
55 pub fn new(ctx: ResearchContext) -> Self {
56 Self { ctx: Arc::new(ctx) }
57 }
58
59 pub async fn serve_stdio(self) -> Result<(), String> {
64 let Some(first_line) = guard::read_until_forwardable().await else {
68 return Ok(());
69 };
70 let stdin = std::io::Cursor::new(first_line).chain(tokio::io::stdin());
72 let service = self
73 .serve((stdin, tokio::io::stdout()))
74 .await
75 .map_err(|e| format!("MCP serve init failed: {e}"))?;
76 service
77 .waiting()
78 .await
79 .map_err(|e| format!("MCP serve stopped: {e}"))?;
80 Ok(())
81 }
82}
83
84fn ok_value(v: Value) -> CallToolResult {
88 CallToolResult::success(vec![rmcp::model::ContentBlock::text(v.to_string())])
89}
90
91fn err_result(e: ResearchError) -> CallToolResult {
94 let kind = match &e {
95 ResearchError::NotFound(_) | ResearchError::Duplicate(_) | ResearchError::Validation(_) => {
96 "invalid_params"
97 }
98 ResearchError::Config(_) => "dependency_missing",
99 ResearchError::Source(_) => "upstream_error",
102 _ => "internal_error", };
104 CallToolResult::error(vec![rmcp::model::ContentBlock::text(format!(
105 "[{kind}] {e}"
106 ))])
107}
108
109macro_rules! tool_result {
111 ($expr:expr) => {
112 match $expr {
113 Ok(v) => ok_value(serde_json::to_value(&v).unwrap_or(Value::Null)),
114 Err(e) => err_result(e),
115 }
116 };
117}
118
119type ToolOutcome<T> = Result<T, CallToolResult>;
123
124async fn ingest_papers(store: &SqliteStore, p: &IngestParams) -> ToolOutcome<(Vec<Paper>, usize)> {
127 if p.source == "pdf" {
128 ingest_pdfs(store, p)
129 } else {
130 ingest_remote(store, p).await
131 }
132}
133
134fn ingest_pdfs(store: &SqliteStore, p: &IngestParams) -> ToolOutcome<(Vec<Paper>, usize)> {
137 let Some(pdf_path) = &p.path else {
138 return Err(err_result(ResearchError::Validation(
139 "path is required for source=pdf".into(),
140 )));
141 };
142 let src = PdfSource::new();
143 let paths = PdfSource::collect_paths(std::path::Path::new(pdf_path)).map_err(err_result)?;
144 let mut papers = Vec::new();
145 let mut skipped = 0usize;
146 for path in &paths {
147 match src.ingest_file(path) {
148 Ok((paper, body)) => {
149 store.insert_paper(&paper).map_err(err_result)?;
150 if let Some(body) = body {
151 store.set_paper_body(&paper.id, &body).map_err(err_result)?;
152 }
153 papers.push(paper);
154 }
155 Err(_) => skipped += 1,
156 }
157 }
158 Ok((papers, skipped))
159}
160
161async fn ingest_remote(store: &SqliteStore, p: &IngestParams) -> ToolOutcome<(Vec<Paper>, usize)> {
164 let Some(q) = &p.query else {
165 return Err(err_result(ResearchError::Validation(format!(
166 "query is required for source={}",
167 p.source
168 ))));
169 };
170 let mut papers = Vec::new();
171 if p.source == "arxiv" || p.source == "all" {
172 let arxiv = ArxivSource::new();
173 let fetched = IngestPipeline::new(&arxiv, store)
174 .run(q, p.limit)
175 .await
176 .map_err(err_result)?;
177 papers.extend(fetched);
178 }
179 if p.source == "s2" || p.source == "all" {
180 let s2 = SemanticScholarSource::new();
181 let fetched = IngestPipeline::new(&s2, store)
182 .run(q, p.limit)
183 .await
184 .map_err(err_result)?;
185 papers.extend(fetched);
186 }
187 if p.source == "openalex" || p.source == "all" {
188 let oa = OpenAlexSource::new();
189 let fetched = IngestPipeline::new(&oa, store)
190 .run(q, p.limit)
191 .await
192 .map_err(err_result)?;
193 papers.extend(fetched);
194 }
195 if p.source == "europepmc" || p.source == "all" {
196 let epmc = EuropePmcSource::new();
197 let fetched = IngestPipeline::new(&epmc, store)
198 .run(q, p.limit)
199 .await
200 .map_err(err_result)?;
201 papers.extend(fetched);
202 }
203 if p.source == "preprints" || p.source == "all" {
204 let pre = PreprintSource::new();
205 let fetched = IngestPipeline::new(&pre, store)
206 .run(q, p.limit)
207 .await
208 .map_err(err_result)?;
209 papers.extend(fetched);
210 }
211 Ok((papers, 0))
212}
213
214fn link_ingested_to_topic(
217 store: &SqliteStore,
218 papers: &[Paper],
219 topic: &Option<String>,
220) -> ToolOutcome<()> {
221 let Some(topic_id) = topic else {
222 return Ok(());
223 };
224 match store.get_topic(topic_id) {
225 Ok(Some(_)) => {}
226 Ok(None) => {
227 return Err(err_result(ResearchError::NotFound(format!(
228 "topic '{topic_id}'"
229 ))));
230 }
231 Err(e) => return Err(err_result(e)),
232 }
233 for paper in papers {
234 store
235 .link_paper_to_topic(&paper.id, topic_id, paper.relevance_score)
236 .map_err(err_result)?;
237 }
238 Ok(())
239}
240
241#[tool_router]
244impl ResearchServer {
245 #[tool(
246 description = "Initialize a research workspace: create the SQLite index schema and default config if missing. Returns the db path."
247 )]
248 pub fn init(&self) -> CallToolResult {
249 let db_path = self.ctx.db_path.clone();
250 if let Err(e) = load_config() {
252 return err_result(e);
253 }
254 match open_store(&db_path) {
255 Ok(store) => match store.init_schema() {
256 Ok(()) => ok_value(json!({
257 "initialized": true,
258 "db": db_path.to_string_lossy(),
259 })),
260 Err(e) => err_result(e),
261 },
262 Err(e) => err_result(e),
263 }
264 }
265
266 #[tool(
267 description = "Ingest papers from arXiv, Semantic Scholar, OpenAlex, Europe PMC (PubMed), bioRxiv-style preprints, or local PDFs. source: arxiv|s2|openalex|europepmc|preprints|all|pdf. For arxiv/s2/openalex/europepmc/preprints/all a query is required; for pdf a path (file or dir) is required. Optionally link ingested papers to a topic. Network-heavy for remote sources (async)."
268 )]
269 pub async fn ingest(&self, Parameters(p): Parameters<IngestParams>) -> CallToolResult {
270 let store = match open_store(&self.ctx.db_path) {
271 Ok(s) => s,
272 Err(e) => return err_result(e),
273 };
274 let (all_papers, skipped) = match ingest_papers(&store, &p).await {
275 Ok((papers, skipped)) => (papers, skipped),
276 Err(resp) => return resp,
277 };
278 if let Err(resp) = link_ingested_to_topic(&store, &all_papers, &p.topic) {
279 return resp;
280 }
281
282 ok_value(json!({
283 "ingested": all_papers.len(),
284 "skipped": skipped,
285 "source": p.source,
286 "linked_topic": p.topic,
287 "papers": serde_json::to_value(&all_papers).unwrap_or(Value::Null),
288 }))
289 }
290
291 #[tool(description = "Force a full rebuild of the FTS search index.")]
292 pub fn index_rebuild(&self) -> CallToolResult {
293 let store = match open_store(&self.ctx.db_path) {
294 Ok(s) => s,
295 Err(e) => return err_result(e),
296 };
297 match store.rebuild_index() {
298 Ok(()) => ok_value(json!({"rebuilt": true})),
299 Err(e) => err_result(e),
300 }
301 }
302
303 #[tool(
304 description = "Import papers from BibTeX/BibLaTeX (.bib) or CSL-JSON (.json) files — e.g. a Zotero export. path is a file or a directory (all matching files are imported). Papers with a DOI already in the library are skipped."
305 )]
306 pub fn import_papers(&self, Parameters(p): Parameters<ImportPapersParams>) -> CallToolResult {
307 let store = match open_store(&self.ctx.db_path) {
308 Ok(s) => s,
309 Err(e) => return err_result(e),
310 };
311 match crate::application::paper_import::run_import(&store, std::path::Path::new(&p.path)) {
312 Ok(summary) => ok_value(json!({
313 "imported": summary.imported.len(),
314 "skipped_duplicates": summary.skipped_duplicates,
315 "failed": summary.failed,
316 "papers": serde_json::to_value(&summary.imported).unwrap_or(Value::Null),
317 })),
318 Err(e) => err_result(e),
319 }
320 }
321
322 #[tool(
323 description = "Fetch the stored full body text of a paper (from PDF ingest), section headings marked with '## ' and page boundaries with '<!-- page N -->'. Pass query to get matching snippets with their section and page instead of the whole body — far cheaper on context. Returns has_body=false if only metadata is stored."
324 )]
325 pub fn paper_body(&self, Parameters(p): Parameters<PaperBodyParams>) -> CallToolResult {
326 let store = match open_store(&self.ctx.db_path) {
327 Ok(s) => s,
328 Err(e) => return err_result(e),
329 };
330 if let Some(query) = p.query.as_deref().filter(|q| !q.trim().is_empty()) {
332 return match store.search_body_evidence(query, Some(&p.id), 20) {
333 Ok(matches) => ok_value(json!({
334 "id": p.id,
335 "query": query,
336 "matches": matches,
337 })),
338 Err(e) => err_result(e),
339 };
340 }
341 match store.get_paper_body(&p.id) {
342 Ok(Some(body)) => {
343 const MAX_TOOL_BODY_CHARS: usize = 40_000;
346 let truncated = body.chars().take(MAX_TOOL_BODY_CHARS).collect::<String>();
347 ok_value(json!({
348 "id": p.id,
349 "has_body": true,
350 "truncated": truncated.chars().count() < body.chars().count(),
351 "text": truncated,
352 }))
353 }
354 Ok(None) => ok_value(json!({ "id": p.id, "has_body": false })),
355 Err(e) => err_result(e),
356 }
357 }
358
359 #[tool(
360 description = "Fetch and store citation-graph edges for a paper via OpenAlex, ingesting newly seen related papers into the library. direction \"references\" (default) lists the works the paper cites; \"cited_by\" lists the works citing it. Set intents=true to additionally label edges with Semantic Scholar citation intents (background/methodology/result, influential); S2 classifies only a fraction of edges, so partial labeling is normal. Requires the paper to have an OpenAlex id or DOI. Idempotent. Network-heavy (async)."
361 )]
362 pub async fn paper_references(
363 &self,
364 Parameters(p): Parameters<PaperReferencesParams>,
365 ) -> CallToolResult {
366 let cited_by = match p.direction.as_deref() {
367 None | Some("references") => false,
368 Some("cited_by") => true,
369 Some(other) => {
370 return err_result(ResearchError::Source(format!(
371 "unknown direction '{other}'; use \"references\" or \"cited_by\""
372 )));
373 }
374 };
375 let store = match open_store(&self.ctx.db_path) {
376 Ok(s) => s,
377 Err(e) => return err_result(e),
378 };
379 let paper = match store.get_paper(&p.id) {
380 Ok(Some(paper)) => paper,
381 Ok(None) => {
382 return err_result(ResearchError::NotFound(format!(
383 "paper '{}' not found",
384 p.id
385 )));
386 }
387 Err(e) => return err_result(e),
388 };
389 let synced = if cited_by {
390 crate::application::references::sync_cited_by(&store, &paper).await
391 } else {
392 crate::application::references::sync_references(&store, &paper).await
393 };
394 match synced {
395 Ok((papers, new_edges, new_papers)) => {
396 let intent_counts = if p.intents.unwrap_or(false) {
401 crate::application::references::sync_citation_intents(&store, &paper, cited_by)
402 .await
403 .ok()
404 } else {
405 None
406 };
407 let edges = if cited_by {
408 store.citations_citing_paper(&p.id)
409 } else {
410 store.citations_for_paper(&p.id)
411 };
412 match edges {
413 Ok(edges) => ok_value(json!({
414 "id": p.id,
415 "direction": if cited_by { "cited_by" } else { "references" },
416 "edges": edges.len(),
417 "new_edges": new_edges,
418 "new_papers_count": new_papers,
419 "intents_labeled": intent_counts.map(|(labeled, _)| labeled),
422 "intents_unlabeled": intent_counts.map(|(_, unlabeled)| unlabeled),
423 "edge_contexts": edges.iter().map(|e| json!({
425 "citing": e.citing_paper_id,
426 "cited": e.cited_paper_id,
427 "context": e.context,
428 })).collect::<Vec<_>>(),
429 "related_papers": papers,
432 })),
433 Err(e) => err_result(e),
434 }
435 }
436 Err(e) => err_result(e),
437 }
438 }
439
440 #[tool(
441 description = "Search the local paper index by query (FTS5 full-text over title, abstract, notes, tags, keywords, and body). Returns matching papers (id, title, authors, year, status). Recall for paraphrased queries depends on stored keywords — see papers_missing_keywords and enrich_paper."
442 )]
443 pub fn query_papers(&self, Parameters(p): Parameters<QueryPapersParams>) -> CallToolResult {
444 let store = match open_store(&self.ctx.db_path) {
445 Ok(s) => s,
446 Err(e) => return err_result(e),
447 };
448 let results = store.search_papers(&p.query, p.limit);
449 tool_result!(results)
450 }
451
452 #[tool(
453 description = "Papers that have no search keywords yet. For each, read its title and abstract, generate 5-10 English keywords (synonyms, expanded acronyms, alternative phrasings a searcher might use that the abstract does not contain), then store them with enrich_paper. This is what makes paraphrased queries findable."
454 )]
455 pub fn papers_missing_keywords(
456 &self,
457 Parameters(p): Parameters<MissingKeywordsParams>,
458 ) -> CallToolResult {
459 let store = match open_store(&self.ctx.db_path) {
460 Ok(s) => s,
461 Err(e) => return err_result(e),
462 };
463 let results = store.papers_missing_keywords(p.limit.min(200));
466 tool_result!(results)
467 }
468
469 #[tool(
470 description = "Store search keywords for one paper, e.g. \"transformer; self-attention; sequence modeling\". Overwrites any previous value and reindexes the paper for full-text search."
471 )]
472 pub fn enrich_paper(&self, Parameters(p): Parameters<EnrichPaperParams>) -> CallToolResult {
473 let store = match open_store(&self.ctx.db_path) {
474 Ok(s) => s,
475 Err(e) => return err_result(e),
476 };
477 match store.set_paper_keywords(&p.id, &p.keywords) {
478 Ok(()) => ok_value(json!({"id": p.id, "keywords": p.keywords})),
479 Err(e) => err_result(e),
480 }
481 }
482
483 #[tool(
484 description = "Collect a topic brief: the topic, its papers with reading status, recorded gaps, and coverage state. Analyze this data yourself, then persist your findings with gaps_record."
485 )]
486 pub fn topic_brief(&self, Parameters(p): Parameters<TopicBriefParams>) -> CallToolResult {
487 let store = match open_store(&self.ctx.db_path) {
488 Ok(s) => s,
489 Err(e) => return err_result(e),
490 };
491 tool_result!(collect_brief(&store, &p.topic))
492 }
493
494 #[tool(
495 description = "Record knowledge gaps you identified for a topic. Each gap is {description, gap_type?, priority?} where gap_type is one of missing_literature | unanswered_question | methodology_gap | connection_gap and priority is 0..=1."
496 )]
497 pub fn gaps_record(&self, Parameters(p): Parameters<GapsRecordParams>) -> CallToolResult {
498 let store = match open_store(&self.ctx.db_path) {
499 Ok(s) => s,
500 Err(e) => return err_result(e),
501 };
502 let gaps: Vec<(String, Option<String>, Option<f32>)> = p
503 .gaps
504 .into_iter()
505 .map(|g| (g.description, g.gap_type, g.priority))
506 .collect();
507 tool_result!(record_gaps(&store, &p.topic, &gaps))
508 }
509
510 #[tool(description = "List recorded knowledge gaps, optionally filtered by topic id.")]
511 pub fn list_gaps(&self, Parameters(p): Parameters<ListGapsParams>) -> CallToolResult {
512 let store = match open_store(&self.ctx.db_path) {
513 Ok(s) => s,
514 Err(e) => return err_result(e),
515 };
516 tool_result!(store.list_gaps(p.topic.as_deref()))
517 }
518
519 #[tool(
520 description = "Collect source material for a report: topic briefs (papers, gaps, coverage) for comma-separated topic ids. Draft the markdown yourself, then store it with report_save."
521 )]
522 pub fn report_material(&self, Parameters(p): Parameters<ReportTopicParams>) -> CallToolResult {
523 let store = match open_store(&self.ctx.db_path) {
524 Ok(s) => s,
525 Err(e) => return err_result(e),
526 };
527 let topic_ids: Vec<String> = p.topics.split(',').map(|t| t.trim().to_string()).collect();
528 tool_result!(collect_material(&store, &topic_ids))
529 }
530
531 #[tool(
532 description = "Store an agent-authored markdown report. Sections are split on '## ' headings. Returns the report id and metadata."
533 )]
534 pub fn report_save(&self, Parameters(p): Parameters<ReportSaveParams>) -> CallToolResult {
535 let store = match open_store(&self.ctx.db_path) {
536 Ok(s) => s,
537 Err(e) => return err_result(e),
538 };
539 let topic_ids: Vec<String> = p.topics.split(',').map(|t| t.trim().to_string()).collect();
540 match save_report(&store, &p.title, &topic_ids, &p.markdown) {
541 Ok(report) => ok_value(json!({
542 "id": report.id,
543 "title": report.title,
544 "sections": report.sections.len(),
545 "markdown": report.to_markdown(),
546 })),
547 Err(e) => err_result(e),
548 }
549 }
550
551 #[tool(description = "List all research topics with their hierarchy depth.")]
552 pub fn topics_list(&self) -> CallToolResult {
553 let store = match open_store(&self.ctx.db_path) {
554 Ok(s) => s,
555 Err(e) => return err_result(e),
556 };
557 tool_result!(store.list_topics())
558 }
559
560 #[tool(description = "Add a research topic, optionally as a sub-topic of a parent.")]
561 pub fn topic_add(&self, Parameters(p): Parameters<TopicAddParams>) -> CallToolResult {
562 let store = match open_store(&self.ctx.db_path) {
563 Ok(s) => s,
564 Err(e) => return err_result(e),
565 };
566 let mut topic = match &p.parent {
567 Some(parent_id) => match store.get_topic(parent_id) {
568 Ok(Some(parent)) => ResearchTopic::new_subtopic(p.name, &parent),
569 Ok(None) => {
570 return err_result(ResearchError::NotFound(format!(
571 "parent topic '{parent_id}'"
572 )));
573 }
574 Err(e) => return err_result(e),
575 },
576 None => ResearchTopic::new(p.name),
577 };
578 topic.description = p.description;
579 let id = topic.id.clone();
580 let depth = topic.depth;
581 match store.insert_topic(&topic) {
582 Ok(()) => ok_value(json!({ "id": id, "depth": depth })),
583 Err(e) => err_result(e),
584 }
585 }
586
587 #[tool(
588 description = "Research state overview: counts of topics/papers/gaps plus per-topic coverage."
589 )]
590 pub fn state(&self) -> CallToolResult {
591 let store = match open_store(&self.ctx.db_path) {
592 Ok(s) => s,
593 Err(e) => return err_result(e),
594 };
595 let topics = match store.list_topics() {
596 Ok(t) => t,
597 Err(e) => return err_result(e),
598 };
599 let papers = match store.list_papers(None) {
600 Ok(p) => p,
601 Err(e) => return err_result(e),
602 };
603 let gaps = match store.list_gaps(None) {
604 Ok(g) => g,
605 Err(e) => return err_result(e),
606 };
607
608 let read = papers
609 .iter()
610 .filter(|p| p.reading_status == ReadingStatus::Completed)
611 .count();
612 let queued = papers
613 .iter()
614 .filter(|p| p.reading_status == ReadingStatus::Queued)
615 .count();
616 let rated = papers.iter().filter(|p| p.rating.is_some()).count();
617
618 let per_topic: Vec<Value> = topics
619 .iter()
620 .map(|t| {
621 let state = store.get_research_state(&t.id).ok().flatten();
622 json!({
623 "name": t.name,
624 "state": state,
625 })
626 })
627 .collect();
628
629 ok_value(json!({
630 "topics": topics.len(),
631 "papers": papers.len(),
632 "gaps": gaps.len(),
633 "read": read,
634 "queued": queued,
635 "rated": rated,
636 "per_topic": per_topic,
637 }))
638 }
639
640 #[tool(
641 description = "Update reading status and/or 1–5 rating of a paper. With neither set, returns the current paper."
642 )]
643 pub fn update_read(&self, Parameters(p): Parameters<UpdateReadParams>) -> CallToolResult {
644 let store = match open_store(&self.ctx.db_path) {
645 Ok(s) => s,
646 Err(e) => return err_result(e),
647 };
648
649 let rating = match p.rating {
651 Some(r) => Some(match Rating::new(r) {
652 Ok(rating) => rating,
653 Err(e) => return err_result(e),
654 }),
655 None => None,
656 };
657
658 if let Some(status) = &p.status
659 && let Err(e) =
660 store.update_reading_status(&p.id, ReadingStatus::from_str_lossy(status))
661 {
662 return err_result(e);
663 }
664 if let Some(rating) = rating
665 && let Err(e) = store.update_rating(&p.id, rating)
666 {
667 return err_result(e);
668 }
669
670 if p.status.is_none() && rating.is_none() {
671 return match store.get_paper(&p.id) {
673 Ok(Some(paper)) => ok_value(serde_json::to_value(&paper).unwrap_or(Value::Null)),
674 Ok(None) => err_result(ResearchError::NotFound(format!("paper '{}'", p.id))),
675 Err(e) => err_result(e),
676 };
677 }
678 ok_value(json!({ "updated": p.id, "status": p.status, "rating": p.rating }))
679 }
680}
681
682#[tool_handler]
683impl ServerHandler for ResearchServer {
684 fn get_info(&self) -> rmcp::model::ServerInfo {
685 let mut info = rmcp::model::ServerInfo::default();
686 info.server_info = rmcp::model::Implementation::new("research", env!("CARGO_PKG_VERSION"));
687 info.capabilities = rmcp::model::ServerCapabilities::builder()
690 .enable_tools()
691 .build();
692 info.instructions = Some(
693 "research-agent: personal academic research memory. Drive the flow: \
694 init, ingest, query_papers, topic_brief, gaps_record, then report_material and report_save. \
695 Organize with topics_list/topic_add, check the overview with state, \
696 and track progress with update_read."
697 .into(),
698 );
699 info
700 }
701}