Job Researcher¶

job_researcher — G6 Researcher job agent.

Cluster: Job Agents | Type: component | MCP Tools: 26

Overview¶

Domain-specialist job agent for academic and applied researchers. Searches the scientific literature (default: arXiv connector), synthesises findings across sources, extracts structured data from papers, compares study methodologies, drafts research sections, and performs corpus-level analysis — all within G6's safety-bounded, audit-trailed JobAgentBlock framework.

When to use:

Conducting a systematic literature search on a research question across arXiv or other corpora
Synthesising findings from multiple papers into a concise, structured narrative review
Extracting tabular data (results, methods, sample sizes) from a set of studies for meta-analysis
Drafting related-work or background sections for a research paper

Pilot-ready, not GA-hardened

job_researcher is suitable for early pilot workflows where a user needs visible literature-search results, explicit failure propagation, and structured research-assistant outputs through MCP or the direct job block. It should not yet be treated as a fully GA-hardened research platform. Before relying on it for paid-user or client-facing production work, verify the actual MCP install/run path in a clean environment, configure the intended persistent job store, validate external literature connectors under expected rate limits, and keep human review in the loop for publication, regulated-domain, or high-stakes research claims.

Reliability envelope: Direct job outputs expose canonical completion_state values (verified, qualified-draft, blocked-escalated) plus warning_card when work is degraded or blocked. MCP responses mirror the same envelope fields.

Capability discovery: The MCP surface exposes 26 tools. get_capabilities reports the connector registry, agentic runtime status, production-gate state, and R2 grounded-QA configuration. list_patterns reports executable vs advisory agentic patterns.

Grounding and fallback: R2 grounded QA is configured with citations required and no standing expert-review requirement for well-grounded analytical work. When the agentic classifier is unavailable, the component degrades to the deterministic keyword floor and surfaces that degradation honestly.

Connector caveat: Real connector search paths may call arXiv, Semantic Scholar, or Crossref. Single-paper fetch is not yet a fully real path for every connector; delegated mock returns are marked with source="mock", degraded=True, and degradation_reason="single_paper_fetch_mock_stub".

Example:

from mvp.job_researcher import JobResearcherBlock, JobResearcherInput

block = JobResearcherBlock()
result = block.infer(JobResearcherInput(
    task="Search the literature on transformer-based models for protein structure prediction and synthesise the top 10 findings",
    context={"source": "arxiv", "domain": "computational-biology", "max_papers": 20},
))
# result.ok → True; result.value → JobResearcherOutput with result, artifacts

Works well with: job_framework, job_scientist, job_analyst

Public API¶

`JobResearcherBlock(JobAgentBlock)`¶

G6 Researcher job agent — delegates to Researcher MCP block.

Field	Type	Default
`name`	`str`	'job_researcher'
`sector`	`SectorClassification`	field(default_factory=lambda: _SECTOR)
`toolkit`	`ToolkitSpec \\| None`	field(default_factory=lambda: JOB_TOOLKITS.get('researcher'))
`mcp_module`	`str`	'mvp.job_researcher.researcher_mcp.server'
`mcp_db_path`	`str`	':memory:'
`capabilities`	`ClassVar[set[type]]`	{Extensible, HumanLearnable, Collaborative, ProblemSolvable, KnowledgeGrounded, Memorable, AgentCommunicable, ExternallyAdaptable}

Methods:

`infer(data: Any)`¶

`JobResearcherInput(JobInput)`¶

Input for the Researcher job agent.

`JobResearcherOutput(JobOutput)`¶

`DataSourceAdapter(ABC)`¶

Abstract base class for research data source adapters.

Methods:

`fetch_sources(query: str, max_results: int = 20) -> list[dict]`¶

Fetch literature sources matching a query.

`fetch_survey_responses(survey_id: str) -> list[dict]`¶

Fetch survey responses for a given survey.

`fetch_interview_transcripts(project_id: str) -> list[dict]`¶

Fetch interview transcripts for a project.

`store_finding(finding: dict) -> str`¶

Store a research finding, return its ID.

`store_report(report: dict) -> str`¶

Store a research report, return its ID.

`InMemoryDataSourceAdapter(DataSourceAdapter)`¶

In-memory data source adapter for testing and lightweight use.

Methods:

`add_sources(sources: list[dict]) -> None`¶

Add literature sources to the in-memory store.

`add_survey_responses(survey_id: str, responses: list[dict]) -> None`¶

Add survey responses for a survey ID.

`add_interview_transcripts(project_id: str, transcripts: list[dict]) -> None`¶

Add interview transcripts for a project.

`fetch_sources(query: str, max_results: int = 20) -> list[dict]`¶

Fetch sources matching query via keyword search.

`fetch_survey_responses(survey_id: str) -> list[dict]`¶

Fetch survey responses by ID.

`fetch_interview_transcripts(project_id: str) -> list[dict]`¶

Fetch interview transcripts by project ID.

`store_finding(finding: dict) -> str`¶

Store a finding and return its ID.

`store_report(report: dict) -> str`¶

Store a report and return its ID.

`get_finding(finding_id: str) -> dict | None`¶

Retrieve a finding by ID.

`get_report(report_id: str) -> dict | None`¶

Retrieve a report by ID.

`list_findings() -> list[dict]`¶

List all stored findings.

`list_reports() -> list[dict]`¶

List all stored reports.

`JobResearcherMCPBlock(AIBlock[MCPJobResearcherInput, MCPJobResearcherOutput, dict])`¶

26-op MCP block for the Researcher job agent.

Field	Type	Default
`name`	`str`	'job_researcher_mcp'
`state`	`dict`	field(default_factory=dict)
`db_path`	`str`	':memory:'
`resource_bounds`	`ResourceBounds`	field(default_factory=ResourceBounds)
`usage`	`ResourceUsage`	field(default_factory=ResourceUsage)

Methods:

`infer(data: MCPJobResearcherInput) -> Result[MCPJobResearcherOutput]`¶

`MCPJobResearcherInput(BaseModel)`¶

Input to JobResearcherMCPBlock — 26-op dispatch.

Field	Type	Default
`op`	`Literal['search_literature', 'synthesize_findings', 'extract_data', 'compare_sources', 'draft_paper', 'analyze_corpus', 'evaluate_sources', 'assess_quality', 'benchmark_methods', 'audit_citations', 'create_proposal', 'review_deliverable', 'delegate_task', 'report_status', 'request_feedback', 'store_artifact', 'retrieve_artifact', 'list_artifacts', 'search_artifacts', 'archive', 'plan_sprint', 'track_progress', 'reflect_on_outcome', 'get_capabilities', 'info', 'list_patterns']`	required
`task`	`str`	''
`context`	`dict[str, Any]`	Field(default_factory=dict)
`parameters`	`dict[str, Any]`	Field(default_factory=dict)
`artifact_id`	`str`	''
`query`	`str`	''
`run_mode`	`str`	'beta'
`reviewer_signature`	`str`	''

`MCPJobResearcherOutput(BaseModel)`¶

Output from JobResearcherMCPBlock.

Field	Type	Default
`op`	`str`	required
`result`	`str`	''
`artifacts`	`list[dict[str, Any]]`	Field(default_factory=list)
`records`	`list[dict[str, Any]]`	Field(default_factory=list)
`message`	`str`	''
`count`	`int`	0
`found`	`bool`	False
`metadata`	`dict[str, Any]`	Field(default_factory=dict)
`degraded`	`bool`	False
`degradation_reason`	`str \\| None`	None
`completion_state`	`Literal['verified', 'qualified-draft', 'blocked-escalated']`	'qualified-draft'
`warning_card`	`dict[str, Any] \\| None`	None

`ResearcherStore(JobStore)`¶

SQLite store for the Researcher job agent.

Constructor:

Parameter	Type	Default
`db_path`	`str`	':memory:'

Methods:

`create_project(title: str, research_type: str = '', methodology: str = '', client: str = '', objectives: list[str] | None = None) -> str`¶

`get_project(project_id: str) -> dict | None`¶

`list_projects(status: str = '', research_type: str = '') -> list[dict]`¶

`update_project_status(project_id: str, status: str) -> bool`¶

`create_publication(title: str, project_id: str = '', authors: list[str] | None = None, journal: str = '', year: int = 0, doi: str = '', abstract: str = '', keywords: list[str] | None = None) -> str`¶

`get_publication(publication_id: str) -> dict | None`¶

`list_publications(project_id: str = '', journal: str = '', status: str = '') -> list[dict]`¶

`update_publication(publication_id: str, citation_count: int | None = None, status: str | None = None) -> bool`¶

`create_dataset(name: str, project_id: str = '', description: str = '', source: str = '', fmt: str = '', record_count: int = 0, variables: list[str] | None = None, quality_score: float = 0.0, license_: str = '', url: str = '') -> str`¶

`get_dataset(dataset_id: str) -> dict | None`¶

`list_datasets(project_id: str = '', source: str = '') -> list[dict]`¶

`delete_dataset(dataset_id: str) -> bool`¶

`create_literature_review(title: str, project_id: str = '', review_type: str = 'narrative', query: str = '', inclusion: dict | None = None, exclusion: dict | None = None) -> str`¶

`get_literature_review(review_id: str) -> dict | None`¶

`list_literature_reviews(project_id: str = '', review_type: str = '', status: str = '') -> list[dict]`¶

`update_literature_review(review_id: str, sources: list[dict] | None = None, themes: list[str] | None = None, summary: str | None = None, total_screened: int | None = None, total_included: int | None = None, status: str | None = None) -> bool`¶

`create_grant(title: str, project_id: str = '', funder: str = '', amount: float = 0.0, currency: str = 'USD', start_date: str = '', end_date: str = '', pi_name: str = '', co_pis: list[str] | None = None, objectives: list[str] | None = None, deliverables: list[str] | None = None) -> str`¶

`get_grant(grant_id: str) -> dict | None`¶

`list_grants(project_id: str = '', funder: str = '', status: str = '') -> list[dict]`¶

`update_grant_status(grant_id: str, status: str) -> bool`¶

Functions¶

`assemble_review_text(output: Any) -> str`¶

Collect the reviewable free text from a JobResearcherOutput (duck-typed).

`assess_researcher_output(output: Any, qa_block: Any | None = None, generate: Any | None = None) -> GroundedRunResult`¶

Run grounded four-valued QA over a researcher output.

MCP Tools¶

Operation	Source
`search_literature`	researcher_mcp
`synthesize_findings`	researcher_mcp
`extract_data`	researcher_mcp
`compare_sources`	researcher_mcp
`draft_paper`	researcher_mcp
`analyze_corpus`	researcher_mcp
`evaluate_sources`	researcher_mcp
`assess_quality`	researcher_mcp
`benchmark_methods`	researcher_mcp
`audit_citations`	researcher_mcp
`create_proposal`	researcher_mcp
`review_deliverable`	researcher_mcp
`delegate_task`	researcher_mcp
`report_status`	researcher_mcp
`request_feedback`	researcher_mcp
`store_artifact`	researcher_mcp
`retrieve_artifact`	researcher_mcp
`list_artifacts`	researcher_mcp
`search_artifacts`	researcher_mcp
`archive`	researcher_mcp
`plan_sprint`	researcher_mcp
`track_progress`	researcher_mcp
`reflect_on_outcome`	researcher_mcp
`get_capabilities`	researcher_mcp
`info`	researcher_mcp
`list_patterns`	researcher_mcp

Job Researcher¶

Overview¶

Public API¶

JobResearcherBlock(JobAgentBlock)¶

infer(data: Any)¶

JobResearcherInput(JobInput)¶

JobResearcherOutput(JobOutput)¶

DataSourceAdapter(ABC)¶

fetch_sources(query: str, max_results: int = 20) -> list[dict]¶

fetch_survey_responses(survey_id: str) -> list[dict]¶

fetch_interview_transcripts(project_id: str) -> list[dict]¶

store_finding(finding: dict) -> str¶

store_report(report: dict) -> str¶

InMemoryDataSourceAdapter(DataSourceAdapter)¶

add_sources(sources: list[dict]) -> None¶

add_survey_responses(survey_id: str, responses: list[dict]) -> None¶

add_interview_transcripts(project_id: str, transcripts: list[dict]) -> None¶

fetch_sources(query: str, max_results: int = 20) -> list[dict]¶

fetch_survey_responses(survey_id: str) -> list[dict]¶

fetch_interview_transcripts(project_id: str) -> list[dict]¶

store_finding(finding: dict) -> str¶

store_report(report: dict) -> str¶

get_finding(finding_id: str) -> dict | None¶

get_report(report_id: str) -> dict | None¶

list_findings() -> list[dict]¶

list_reports() -> list[dict]¶

JobResearcherMCPBlock(AIBlock[MCPJobResearcherInput, MCPJobResearcherOutput, dict])¶

infer(data: MCPJobResearcherInput) -> Result[MCPJobResearcherOutput]¶

MCPJobResearcherInput(BaseModel)¶

MCPJobResearcherOutput(BaseModel)¶

ResearcherStore(JobStore)¶

create_project(title: str, research_type: str = '', methodology: str = '', client: str = '', objectives: list[str] | None = None) -> str¶

get_project(project_id: str) -> dict | None¶

list_projects(status: str = '', research_type: str = '') -> list[dict]¶

update_project_status(project_id: str, status: str) -> bool¶

create_publication(title: str, project_id: str = '', authors: list[str] | None = None, journal: str = '', year: int = 0, doi: str = '', abstract: str = '', keywords: list[str] | None = None) -> str¶

get_publication(publication_id: str) -> dict | None¶

list_publications(project_id: str = '', journal: str = '', status: str = '') -> list[dict]¶

update_publication(publication_id: str, citation_count: int | None = None, status: str | None = None) -> bool¶

create_dataset(name: str, project_id: str = '', description: str = '', source: str = '', fmt: str = '', record_count: int = 0, variables: list[str] | None = None, quality_score: float = 0.0, license_: str = '', url: str = '') -> str¶

get_dataset(dataset_id: str) -> dict | None¶

list_datasets(project_id: str = '', source: str = '') -> list[dict]¶

delete_dataset(dataset_id: str) -> bool¶

create_literature_review(title: str, project_id: str = '', review_type: str = 'narrative', query: str = '', inclusion: dict | None = None, exclusion: dict | None = None) -> str¶

get_literature_review(review_id: str) -> dict | None¶

list_literature_reviews(project_id: str = '', review_type: str = '', status: str = '') -> list[dict]¶

update_literature_review(review_id: str, sources: list[dict] | None = None, themes: list[str] | None = None, summary: str | None = None, total_screened: int | None = None, total_included: int | None = None, status: str | None = None) -> bool¶

create_grant(title: str, project_id: str = '', funder: str = '', amount: float = 0.0, currency: str = 'USD', start_date: str = '', end_date: str = '', pi_name: str = '', co_pis: list[str] | None = None, objectives: list[str] | None = None, deliverables: list[str] | None = None) -> str¶

get_grant(grant_id: str) -> dict | None¶

list_grants(project_id: str = '', funder: str = '', status: str = '') -> list[dict]¶

update_grant_status(grant_id: str, status: str) -> bool¶

Functions¶

assemble_review_text(output: Any) -> str¶

assess_researcher_output(output: Any, qa_block: Any | None = None, generate: Any | None = None) -> GroundedRunResult¶

MCP Tools¶

`JobResearcherBlock(JobAgentBlock)`¶

`infer(data: Any)`¶

`JobResearcherInput(JobInput)`¶

`JobResearcherOutput(JobOutput)`¶

`DataSourceAdapter(ABC)`¶

`fetch_sources(query: str, max_results: int = 20) -> list[dict]`¶

`fetch_survey_responses(survey_id: str) -> list[dict]`¶

`fetch_interview_transcripts(project_id: str) -> list[dict]`¶

`store_finding(finding: dict) -> str`¶

`store_report(report: dict) -> str`¶

`InMemoryDataSourceAdapter(DataSourceAdapter)`¶

`add_sources(sources: list[dict]) -> None`¶

`add_survey_responses(survey_id: str, responses: list[dict]) -> None`¶

`add_interview_transcripts(project_id: str, transcripts: list[dict]) -> None`¶

`fetch_sources(query: str, max_results: int = 20) -> list[dict]`¶

`fetch_survey_responses(survey_id: str) -> list[dict]`¶

`fetch_interview_transcripts(project_id: str) -> list[dict]`¶

`store_finding(finding: dict) -> str`¶

`store_report(report: dict) -> str`¶

`get_finding(finding_id: str) -> dict | None`¶

`get_report(report_id: str) -> dict | None`¶

`list_findings() -> list[dict]`¶

`list_reports() -> list[dict]`¶

`JobResearcherMCPBlock(AIBlock[MCPJobResearcherInput, MCPJobResearcherOutput, dict])`¶

`infer(data: MCPJobResearcherInput) -> Result[MCPJobResearcherOutput]`¶

`MCPJobResearcherInput(BaseModel)`¶

`MCPJobResearcherOutput(BaseModel)`¶

`ResearcherStore(JobStore)`¶

`create_project(title: str, research_type: str = '', methodology: str = '', client: str = '', objectives: list[str] | None = None) -> str`¶

`get_project(project_id: str) -> dict | None`¶

`list_projects(status: str = '', research_type: str = '') -> list[dict]`¶

`update_project_status(project_id: str, status: str) -> bool`¶

`create_publication(title: str, project_id: str = '', authors: list[str] | None = None, journal: str = '', year: int = 0, doi: str = '', abstract: str = '', keywords: list[str] | None = None) -> str`¶

`get_publication(publication_id: str) -> dict | None`¶

`list_publications(project_id: str = '', journal: str = '', status: str = '') -> list[dict]`¶

`update_publication(publication_id: str, citation_count: int | None = None, status: str | None = None) -> bool`¶

`create_dataset(name: str, project_id: str = '', description: str = '', source: str = '', fmt: str = '', record_count: int = 0, variables: list[str] | None = None, quality_score: float = 0.0, license_: str = '', url: str = '') -> str`¶

`get_dataset(dataset_id: str) -> dict | None`¶

`list_datasets(project_id: str = '', source: str = '') -> list[dict]`¶

`delete_dataset(dataset_id: str) -> bool`¶

`create_literature_review(title: str, project_id: str = '', review_type: str = 'narrative', query: str = '', inclusion: dict | None = None, exclusion: dict | None = None) -> str`¶

`get_literature_review(review_id: str) -> dict | None`¶

`list_literature_reviews(project_id: str = '', review_type: str = '', status: str = '') -> list[dict]`¶

`update_literature_review(review_id: str, sources: list[dict] | None = None, themes: list[str] | None = None, summary: str | None = None, total_screened: int | None = None, total_included: int | None = None, status: str | None = None) -> bool`¶

`create_grant(title: str, project_id: str = '', funder: str = '', amount: float = 0.0, currency: str = 'USD', start_date: str = '', end_date: str = '', pi_name: str = '', co_pis: list[str] | None = None, objectives: list[str] | None = None, deliverables: list[str] | None = None) -> str`¶

`get_grant(grant_id: str) -> dict | None`¶

`list_grants(project_id: str = '', funder: str = '', status: str = '') -> list[dict]`¶

`update_grant_status(grant_id: str, status: str) -> bool`¶

`assemble_review_text(output: Any) -> str`¶

`assess_researcher_output(output: Any, qa_block: Any | None = None, generate: Any | None = None) -> GroundedRunResult`¶