Job Scientist¶

job_scientist -- G6 Scientist job agent.

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

Overview¶

Domain-specialist job agent for research scientists across natural and applied sciences. Designs experiments with appropriate controls, performs statistical data analysis, conducts systematic literature reviews, tests and refines hypotheses, and drafts publication-ready findings — all within G6's safety-bounded, audit-trailed JobAgentBlock framework.

When to use:

Designing a controlled experiment with sample-size calculations and randomisation scheme
Performing statistical analysis (ANOVA, regression, Bayesian inference) on experimental data
Conducting a rigorous literature review to position a hypothesis within existing knowledge
Drafting results and discussion sections for a peer-reviewed manuscript

Example:

from mvp.job_scientist import JobScientistBlock, JobScientistInput

block = JobScientistBlock()
result = block.infer(JobScientistInput(
    task="Design a randomised controlled trial to test whether daily magnesium supplementation reduces migraine frequency in adults",
    context={"population": "adults 18-65 with episodic migraine", "duration_weeks": 12, "alpha": 0.05, "power": 0.80},
))
# result.ok → True; result.value → JobScientistOutput with result, artifacts

Works well with: job_framework, job_researcher, job_analyst

Literature review and live-source caveats

job_scientist keeps first-run MCP workflows fast and deterministic. Literature review uses Semantic Scholar when live retrieval is available, but live failures, empty responses, rate limits, or offline environments degrade to labelled fallback papers unless G6_SCIENTIST_STRICT_SEMANTIC_SCHOLAR=true is set. Check metadata.connector_status, each paper's connector_status, and metadata.synthesis_details.evidence_profile.source_mode before treating retrieved literature as live evidence.

The built-in synthesis is metadata-level only. It summarises paper count, year range, citation counts, notable papers, limitations, and next steps from available metadata; it is not a full-text systematic review and should not be used as scientific, clinical, regulatory, or publication sign-off without human review.

Optional enrichment integrations are bounded and opt-in. Enable them only when their latency and dependency risk are acceptable: G6_SCIENTIST_ENABLE_GROUNDING, G6_SCIENTIST_ENABLE_WEB_SEARCH, G6_SCIENTIST_ENABLE_DEBATE, or G6_SCIENTIST_ENABLE_DEEP_INTEGRATIONS. Time budgets can be tuned with G6_SCIENTIST_<NAME>_TIMEOUT_SEC or G6_SCIENTIST_INTEGRATION_TIMEOUT_SEC.

Public API¶

`ScientistDecision`¶

Validated op-classification decision for a scientific task.

Field	Type	Default
`op`	`str`	required
`recommendation`	`str`	''
`rationale`	`str`	''
`signals`	`list[str]`	field(default_factory=list)
`citations`	`list[str]`	field(default_factory=list)
`confidence`	`float`	0.0
`completion_state`	`str`	'qualified-draft'
`degraded`	`bool`	False
`requires_human_review`	`bool`	False
`raw_response`	`str`	''

`ScientistPlanner`¶

Runtime-first facade with deterministic fallback (ACTION tier).

Constructor:

Parameter	Type	Default
`runtime`	`ScientistRuntime \\| None`	None

Methods:

`classify(task: str, context: dict | None = None) -> ScientistDecision`¶

`JobScientistInput(JobInput)`¶

Input for the Scientist job agent.

`JobScientistOutput(JobOutput)`¶

`JobScientistBlock(JobAgentBlock)`¶

G6 Scientist job agent — Tier 1 block with MCP delegation and one safety floor.

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

`LiteratureDatabaseAdapter(ABC)`¶

Abstract base class for literature database access.

Methods:

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

Search for papers matching a query string.

`get_paper(paper_id: str) -> dict[str, Any] | None`¶

Retrieve a single paper by ID.

`get_citations(paper_id: str) -> list[dict[str, Any]]`¶

Get papers that cite the given paper.

`get_references(paper_id: str) -> list[dict[str, Any]]`¶

Get papers referenced by the given paper.

`add_paper(paper: dict[str, Any]) -> str`¶

Add a paper to the database. Returns paper_id.

`count() -> int`¶

Return total number of papers in the database.

`InMemoryLiteratureAdapter(LiteratureDatabaseAdapter)`¶

In-memory literature database for testing and lightweight usage.

Methods:

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

Search papers by title/abstract keyword matching.

`get_paper(paper_id: str) -> dict[str, Any] | None`¶

`get_citations(paper_id: str) -> list[dict[str, Any]]`¶

`get_references(paper_id: str) -> list[dict[str, Any]]`¶

`add_paper(paper: dict[str, Any]) -> str`¶

`count() -> int`¶

`remove_paper(paper_id: str) -> bool`¶

Remove a paper and its citation/reference links.

`list_all(limit: int = 100) -> list[dict[str, Any]]`¶

List all papers, sorted by created_at descending.

`MCPJobScientistInput(BaseModel)`¶

Input to JobScientistMCPBlock — 26-op dispatch.

Field	Type	Default
`op`	`Literal['design_experiment', 'analyze_data', 'literature_review', 'test_hypothesis', 'publish_findings', 'analyze_results', 'evaluate_methodology', 'assess_significance', 'benchmark_model', 'audit_reproducibility', '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`	''

`MCPJobScientistOutput(BaseModel)`¶

Output from JobScientistMCPBlock.

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'
`human_review_required`	`bool`	False

`JobScientistMCPBlock(AIBlock[MCPJobScientistInput, MCPJobScientistOutput, dict])`¶

26-op MCP block for the Scientist job agent.

Field	Type	Default
`name`	`str`	'job_scientist_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: MCPJobScientistInput) -> Result[MCPJobScientistOutput]`¶

`ScientistStore(JobStore)`¶

SQLite store for the Scientist job agent.

Constructor:

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

Methods:

`create_experiment(title: str, hypothesis: str = '', methodology: str = '', variables: dict | None = None, status: str = 'planned') -> dict`¶

`get_experiment(experiment_id: str) -> dict | None`¶

`list_experiments(status: str | None = None) -> list[dict]`¶

`update_experiment(experiment_id: str, **fields: Any) -> bool`¶

`create_dataset(name: str, experiment_id: str = '', source: str = '', fmt: str = '', size: int = 0, schema: dict | None = None, data: dict | None = None) -> dict`¶

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

`list_datasets(experiment_id: str | None = None) -> list[dict]`¶

`create_publication(title: str, authors: list[str] | None = None, journal: str = '', doi: str = '', abstract: str = '', citations: int = 0, status: str = 'draft') -> dict`¶

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

`list_publications(status: str | None = None) -> list[dict]`¶

`create_grant(title: str, agency: str = '', amount: float = 0.0, pi: str = '', co_pis: list[str] | None = None, status: str = 'draft', data: dict | None = None) -> dict`¶

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

`list_grants(status: str | None = None) -> list[dict]`¶

`create_literature_review(topic: str, search_terms: list[str] | None = None, papers: list[dict] | None = None, synthesis: str = '', methodology: str = '') -> dict`¶

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

`list_literature_reviews(topic: str | None = None) -> list[dict]`¶

Functions¶

`summarize_job_scientist_agentic_evidence(decisions: list[dict[str, Any]]) -> dict[str, Any]`¶

Summarise runtime-vs-fallback scientist decisions with path redaction.

`agentic_planner_enabled(default_enabled: bool) -> bool`¶

Decide whether the agentic scientist planner should be used.

`deterministic_classify(task: str, context: dict | None = None) -> ScientistDecision`¶

Demoted real keyword op-classifier -- the honest offline floor.

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

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

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

Run grounded four-valued QA over a scientist output.

`applied_agentic_patterns() -> list[dict[str, Any]]`¶

Return compact metadata for the scientist-applied vendored patterns.

`applied_agentic_patterns_summary() -> dict[str, Any]`¶

Compact roll-up used by the list_patterns handler / skill catalog.

`get_skill_catalog() -> ScientistSkillCatalog`¶

`run_research_cycle(store: ScientistStore, parameters: dict[str, Any] | None = None) -> dict[str, Any]`¶

Execute a full research cycle: lit review -> hypothesis -> design -> analysis -> publish.

MCP Tools¶

Operation	Source
`design_experiment`	scientist_mcp
`analyze_data`	scientist_mcp
`literature_review`	scientist_mcp
`test_hypothesis`	scientist_mcp
`publish_findings`	scientist_mcp
`analyze_results`	scientist_mcp
`evaluate_methodology`	scientist_mcp
`assess_significance`	scientist_mcp
`benchmark_model`	scientist_mcp
`audit_reproducibility`	scientist_mcp
`create_proposal`	scientist_mcp
`review_deliverable`	scientist_mcp
`delegate_task`	scientist_mcp
`report_status`	scientist_mcp
`request_feedback`	scientist_mcp
`store_artifact`	scientist_mcp
`retrieve_artifact`	scientist_mcp
`list_artifacts`	scientist_mcp
`search_artifacts`	scientist_mcp
`archive`	scientist_mcp
`plan_sprint`	scientist_mcp
`track_progress`	scientist_mcp
`reflect_on_outcome`	scientist_mcp
`get_capabilities`	scientist_mcp
`info`	scientist_mcp
`list_patterns`	scientist_mcp

Job Scientist¶

Overview¶

Public API¶

ScientistDecision¶

ScientistPlanner¶

classify(task: str, context: dict | None = None) -> ScientistDecision¶

JobScientistInput(JobInput)¶

JobScientistOutput(JobOutput)¶

JobScientistBlock(JobAgentBlock)¶

LiteratureDatabaseAdapter(ABC)¶

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

get_paper(paper_id: str) -> dict[str, Any] | None¶

get_citations(paper_id: str) -> list[dict[str, Any]]¶

get_references(paper_id: str) -> list[dict[str, Any]]¶

add_paper(paper: dict[str, Any]) -> str¶

count() -> int¶

InMemoryLiteratureAdapter(LiteratureDatabaseAdapter)¶

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

get_paper(paper_id: str) -> dict[str, Any] | None¶

get_citations(paper_id: str) -> list[dict[str, Any]]¶

get_references(paper_id: str) -> list[dict[str, Any]]¶

add_paper(paper: dict[str, Any]) -> str¶

count() -> int¶

remove_paper(paper_id: str) -> bool¶

list_all(limit: int = 100) -> list[dict[str, Any]]¶

MCPJobScientistInput(BaseModel)¶

MCPJobScientistOutput(BaseModel)¶

JobScientistMCPBlock(AIBlock[MCPJobScientistInput, MCPJobScientistOutput, dict])¶

infer(data: MCPJobScientistInput) -> Result[MCPJobScientistOutput]¶

ScientistStore(JobStore)¶

create_experiment(title: str, hypothesis: str = '', methodology: str = '', variables: dict | None = None, status: str = 'planned') -> dict¶

get_experiment(experiment_id: str) -> dict | None¶

list_experiments(status: str | None = None) -> list[dict]¶

update_experiment(experiment_id: str, **fields: Any) -> bool¶

create_dataset(name: str, experiment_id: str = '', source: str = '', fmt: str = '', size: int = 0, schema: dict | None = None, data: dict | None = None) -> dict¶

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

list_datasets(experiment_id: str | None = None) -> list[dict]¶

create_publication(title: str, authors: list[str] | None = None, journal: str = '', doi: str = '', abstract: str = '', citations: int = 0, status: str = 'draft') -> dict¶

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

list_publications(status: str | None = None) -> list[dict]¶

create_grant(title: str, agency: str = '', amount: float = 0.0, pi: str = '', co_pis: list[str] | None = None, status: str = 'draft', data: dict | None = None) -> dict¶

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

list_grants(status: str | None = None) -> list[dict]¶

create_literature_review(topic: str, search_terms: list[str] | None = None, papers: list[dict] | None = None, synthesis: str = '', methodology: str = '') -> dict¶

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

list_literature_reviews(topic: str | None = None) -> list[dict]¶

Functions¶

summarize_job_scientist_agentic_evidence(decisions: list[dict[str, Any]]) -> dict[str, Any]¶

agentic_planner_enabled(default_enabled: bool) -> bool¶

deterministic_classify(task: str, context: dict | None = None) -> ScientistDecision¶

assemble_review_text(output: Any) -> str¶

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

applied_agentic_patterns() -> list[dict[str, Any]]¶

applied_agentic_patterns_summary() -> dict[str, Any]¶

get_skill_catalog() -> ScientistSkillCatalog¶

run_research_cycle(store: ScientistStore, parameters: dict[str, Any] | None = None) -> dict[str, Any]¶

MCP Tools¶

`ScientistDecision`¶

`ScientistPlanner`¶

`classify(task: str, context: dict | None = None) -> ScientistDecision`¶

`JobScientistInput(JobInput)`¶

`JobScientistOutput(JobOutput)`¶

`JobScientistBlock(JobAgentBlock)`¶

`LiteratureDatabaseAdapter(ABC)`¶

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

`get_paper(paper_id: str) -> dict[str, Any] | None`¶

`get_citations(paper_id: str) -> list[dict[str, Any]]`¶

`get_references(paper_id: str) -> list[dict[str, Any]]`¶

`add_paper(paper: dict[str, Any]) -> str`¶

`count() -> int`¶

`InMemoryLiteratureAdapter(LiteratureDatabaseAdapter)`¶

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

`get_paper(paper_id: str) -> dict[str, Any] | None`¶

`get_citations(paper_id: str) -> list[dict[str, Any]]`¶

`get_references(paper_id: str) -> list[dict[str, Any]]`¶

`add_paper(paper: dict[str, Any]) -> str`¶

`count() -> int`¶

`remove_paper(paper_id: str) -> bool`¶

`list_all(limit: int = 100) -> list[dict[str, Any]]`¶

`MCPJobScientistInput(BaseModel)`¶

`MCPJobScientistOutput(BaseModel)`¶

`JobScientistMCPBlock(AIBlock[MCPJobScientistInput, MCPJobScientistOutput, dict])`¶

`infer(data: MCPJobScientistInput) -> Result[MCPJobScientistOutput]`¶

`ScientistStore(JobStore)`¶

`create_experiment(title: str, hypothesis: str = '', methodology: str = '', variables: dict | None = None, status: str = 'planned') -> dict`¶

`get_experiment(experiment_id: str) -> dict | None`¶

`list_experiments(status: str | None = None) -> list[dict]`¶

`update_experiment(experiment_id: str, **fields: Any) -> bool`¶

`create_dataset(name: str, experiment_id: str = '', source: str = '', fmt: str = '', size: int = 0, schema: dict | None = None, data: dict | None = None) -> dict`¶

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

`list_datasets(experiment_id: str | None = None) -> list[dict]`¶

`create_publication(title: str, authors: list[str] | None = None, journal: str = '', doi: str = '', abstract: str = '', citations: int = 0, status: str = 'draft') -> dict`¶

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

`list_publications(status: str | None = None) -> list[dict]`¶

`create_grant(title: str, agency: str = '', amount: float = 0.0, pi: str = '', co_pis: list[str] | None = None, status: str = 'draft', data: dict | None = None) -> dict`¶

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

`list_grants(status: str | None = None) -> list[dict]`¶

`create_literature_review(topic: str, search_terms: list[str] | None = None, papers: list[dict] | None = None, synthesis: str = '', methodology: str = '') -> dict`¶

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

`list_literature_reviews(topic: str | None = None) -> list[dict]`¶

`summarize_job_scientist_agentic_evidence(decisions: list[dict[str, Any]]) -> dict[str, Any]`¶

`agentic_planner_enabled(default_enabled: bool) -> bool`¶

`deterministic_classify(task: str, context: dict | None = None) -> ScientistDecision`¶

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

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

`applied_agentic_patterns() -> list[dict[str, Any]]`¶

`applied_agentic_patterns_summary() -> dict[str, Any]`¶

`get_skill_catalog() -> ScientistSkillCatalog`¶

`run_research_cycle(store: ScientistStore, parameters: dict[str, Any] | None = None) -> dict[str, Any]`¶