Job Analyst¶

job_analyst — G6 Analyst job agent.

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

Overview¶

Domain-specialist job agent for data and business analysts. It queries caller-supplied or persisted datasets, builds deterministic regression/forecast/trend models, visualises trends, segments cohorts, and generates arithmetic/statistical insights within G6's safety-bounded, audit-trailed JobAgentBlock framework.

It does not execute live SQL/data-warehouse connectors, scikit-learn training, or AutoML backends. Callers must provide inline data lists or previously saved dataset IDs. build_model uses deterministic analyst rules, not unwired ML/AutoML integrations.

When to use:

Querying structured datasets and surfacing KPI trends for stakeholder reports
Segmenting customer or user cohorts for targeted strategy recommendations
Building lightweight deterministic predictive models from historical business data
Automating recurring analytical reports with version-tracked artifact storage
Discovering the real MCP surface with get_capabilities or pattern coverage with list_patterns
Running caller-invoked post-hoc grounded QA via assess_analyst_output

Reliability envelope:

completion_state is one of verified, qualified-draft, or blocked-escalated
warning_card is populated for degraded planner fallback or advisory human review
evidence records whether capability metadata came from code introspection or runtime execution
Sensitive ops (evaluate_model, assess_accuracy, benchmark_prediction, audit_pipeline) set human_review_required=True; this is advisory at the direct MCP boundary and does not block Tier-2 execution

Grounded QA:

mvp.job_analyst.grounded_qa.assess_analyst_output is a decoupled R2 post-hoc QA utility. Invoke it when an analyst output needs citation/standards checking; if a future caller integrates the verdict inline, map the verdict into completion_state, warning_card, and evidence rather than hiding it in prose.

Example:

from mvp.job_analyst import JobAnalystBlock, JobAnalystInput

block = JobAnalystBlock()
result = block.infer(JobAnalystInput(
    task="Segment active customers by purchase frequency and average order value for Q1 2026",
    parameters={"data": [{"customer": "A", "frequency": 5, "monetary": 120.0}]},
))
# result.ok -> True; result.value -> JobAnalystOutput with completion_state, warning_card, evidence

Works well with: job_framework, job_business, job_finance

Public API¶

`JobAnalystBlock(JobAgentBlock)`¶

G6 Analyst job agent — Tier 1 block with MCP delegation.

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

Methods:

`infer(data: JobAnalystInput) -> Result[JobAnalystOutput]`¶

`JobAnalystInput(JobInput)`¶

Input for the Analyst job agent.

`JobAnalystOutput(JobOutput)`¶

`JobAnalystMCPBlock(AIBlock[MCPJobAnalystInput, MCPJobAnalystOutput, dict])`¶

26-op MCP block for the Analyst job agent.

Field	Type	Default
`name`	`str`	'job_analyst_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: MCPJobAnalystInput) -> Result[MCPJobAnalystOutput]`¶

`MCPJobAnalystInput(BaseModel)`¶

Input to JobAnalystMCPBlock — 26-op dispatch.

Field	Type	Default
`op`	`Literal['query_data', 'build_model', 'visualize_trend', 'segment_cohort', 'generate_insight', 'analyze_dataset', 'evaluate_model', 'assess_accuracy', 'benchmark_prediction', 'audit_pipeline', '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', 'list_patterns', 'get_capabilities', 'info']`	required
`task`	`str`	''
`context`	`dict[str, Any]`	Field(default_factory=dict)
`parameters`	`dict[str, Any]`	Field(default_factory=dict)
`artifact_id`	`str`	''
`query`	`str`	''

`MCPJobAnalystOutput(BaseModel)`¶

Output from JobAnalystMCPBlock.

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]`	Field(default_factory=dict)
`evidence`	`dict[str, Any]`	Field(default_factory=dict)
`human_review_required`	`bool`	False
`review_status`	`Literal['', 'pending', 'reviewed']`	''
`request_id`	`str`	''
`task_id`	`str`	''
`run_id`	`str`	''

`AnalystStore(JobStore)`¶

SQLite store for the Analyst job agent.

Constructor:

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

Methods:

`save_dataset(name: str, data: list[dict], source: str = '', schema: dict | None = None) -> str`¶

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

`list_datasets(status: str = 'active') -> list[dict]`¶

`save_analysis(dataset_id: str, analysis_type: str, methodology: str, results: dict, conclusions: str = '') -> str`¶

`get_analysis(analysis_id: str) -> dict | None`¶

`list_analyses(dataset_id: str = '') -> list[dict]`¶

`save_report(analysis_id: str, title: str, report_type: str = 'summary', sections: list[dict] | None = None) -> str`¶

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

`list_reports(status: str = '') -> list[dict]`¶

`save_requirement(project_name: str, description: str, stakeholder: str = '', priority: str = 'medium', acceptance_criteria: list[str] | None = None) -> str`¶

`get_requirement(req_id: str) -> dict | None`¶

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

`save_kpi(name: str, category: str = '', target: float = 0, current_value: float = 0, trend: str = 'flat', data: list | None = None) -> str`¶

`get_kpi(kpi_id: str) -> dict | None`¶

`list_kpis(category: str = '') -> list[dict]`¶

`update_kpi(kpi_id: str, current_value: float, trend: str = '') -> bool`¶

Functions¶

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

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

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

Run grounded four-valued QA over an analyst output.

MCP Tools¶

Operation	Source
`query_data`	analyst_mcp
`build_model`	analyst_mcp
`visualize_trend`	analyst_mcp
`segment_cohort`	analyst_mcp
`generate_insight`	analyst_mcp
`analyze_dataset`	analyst_mcp
`evaluate_model`	analyst_mcp
`assess_accuracy`	analyst_mcp
`benchmark_prediction`	analyst_mcp
`audit_pipeline`	analyst_mcp
`create_proposal`	analyst_mcp
`review_deliverable`	analyst_mcp
`delegate_task`	analyst_mcp
`report_status`	analyst_mcp
`request_feedback`	analyst_mcp
`store_artifact`	analyst_mcp
`retrieve_artifact`	analyst_mcp
`list_artifacts`	analyst_mcp
`search_artifacts`	analyst_mcp
`archive`	analyst_mcp
`plan_sprint`	analyst_mcp
`track_progress`	analyst_mcp
`reflect_on_outcome`	analyst_mcp
`list_patterns`	analyst_mcp
`get_capabilities`	analyst_mcp
`info`	analyst_mcp
``	analyst_mcp
`pending`	analyst_mcp
`reviewed`	analyst_mcp

Job Analyst¶

Overview¶

Public API¶

JobAnalystBlock(JobAgentBlock)¶

infer(data: JobAnalystInput) -> Result[JobAnalystOutput]¶

JobAnalystInput(JobInput)¶

JobAnalystOutput(JobOutput)¶

JobAnalystMCPBlock(AIBlock[MCPJobAnalystInput, MCPJobAnalystOutput, dict])¶

infer(data: MCPJobAnalystInput) -> Result[MCPJobAnalystOutput]¶

MCPJobAnalystInput(BaseModel)¶

MCPJobAnalystOutput(BaseModel)¶

AnalystStore(JobStore)¶

save_dataset(name: str, data: list[dict], source: str = '', schema: dict | None = None) -> str¶

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

list_datasets(status: str = 'active') -> list[dict]¶

save_analysis(dataset_id: str, analysis_type: str, methodology: str, results: dict, conclusions: str = '') -> str¶

get_analysis(analysis_id: str) -> dict | None¶

list_analyses(dataset_id: str = '') -> list[dict]¶

save_report(analysis_id: str, title: str, report_type: str = 'summary', sections: list[dict] | None = None) -> str¶

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

list_reports(status: str = '') -> list[dict]¶

save_requirement(project_name: str, description: str, stakeholder: str = '', priority: str = 'medium', acceptance_criteria: list[str] | None = None) -> str¶

get_requirement(req_id: str) -> dict | None¶

list_requirements(project_name: str = '', status: str = '') -> list[dict]¶

save_kpi(name: str, category: str = '', target: float = 0, current_value: float = 0, trend: str = 'flat', data: list | None = None) -> str¶

get_kpi(kpi_id: str) -> dict | None¶

list_kpis(category: str = '') -> list[dict]¶

update_kpi(kpi_id: str, current_value: float, trend: str = '') -> bool¶

Functions¶

assemble_review_text(output: Any) -> str¶

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

MCP Tools¶

`JobAnalystBlock(JobAgentBlock)`¶

`infer(data: JobAnalystInput) -> Result[JobAnalystOutput]`¶

`JobAnalystInput(JobInput)`¶

`JobAnalystOutput(JobOutput)`¶

`JobAnalystMCPBlock(AIBlock[MCPJobAnalystInput, MCPJobAnalystOutput, dict])`¶

`infer(data: MCPJobAnalystInput) -> Result[MCPJobAnalystOutput]`¶

`MCPJobAnalystInput(BaseModel)`¶

`MCPJobAnalystOutput(BaseModel)`¶

`AnalystStore(JobStore)`¶

`save_dataset(name: str, data: list[dict], source: str = '', schema: dict | None = None) -> str`¶

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

`list_datasets(status: str = 'active') -> list[dict]`¶

`save_analysis(dataset_id: str, analysis_type: str, methodology: str, results: dict, conclusions: str = '') -> str`¶

`get_analysis(analysis_id: str) -> dict | None`¶

`list_analyses(dataset_id: str = '') -> list[dict]`¶

`save_report(analysis_id: str, title: str, report_type: str = 'summary', sections: list[dict] | None = None) -> str`¶

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

`list_reports(status: str = '') -> list[dict]`¶

`save_requirement(project_name: str, description: str, stakeholder: str = '', priority: str = 'medium', acceptance_criteria: list[str] | None = None) -> str`¶

`get_requirement(req_id: str) -> dict | None`¶

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

`save_kpi(name: str, category: str = '', target: float = 0, current_value: float = 0, trend: str = 'flat', data: list | None = None) -> str`¶

`get_kpi(kpi_id: str) -> dict | None`¶

`list_kpis(category: str = '') -> list[dict]`¶

`update_kpi(kpi_id: str, current_value: float, trend: str = '') -> bool`¶

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

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