Job Consultant¶

job_consultant — G6 Consultant job agent.

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

Overview¶

Domain-specialist job agent for management and strategy consultants. Assesses client situations, develops strategic options, prepares polished deliverable reports, delivers recommendations, conducts post-engagement reviews, and runs competitive market analyses — providing a structured consulting workflow within G6's safety-bounded, audit-trailed JobAgentBlock framework.

When to use:

Conducting a rapid situation assessment and producing a structured findings report for a client
Developing strategic options with pros/cons and implementation roadmaps
Preparing executive-level slide decks or written deliverables from raw analysis
Running post-engagement retrospectives and capturing lessons learned

Example:

from mvp.job_consultant import JobConsultantBlock, JobConsultantInput

block = JobConsultantBlock()
result = block.infer(JobConsultantInput(
    task="Assess the operational inefficiencies in a mid-market logistics firm and recommend top 3 improvements",
    context={"client": "FastFreight Pty Ltd", "engagement_days": 10},
))
# result.ok → True; result.value → JobConsultantOutput with result, artifacts

Optional enrichment:

Consultant tools are deterministic and fast by default. For research-backed outputs, pass integration controls through parameters or context:

result = block.infer(JobConsultantInput(
    task="Benchmark onboarding performance for a new self-serve SaaS product",
    parameters={
        "integration_preset": "enriched",
        "integration_timeout_sec": 2.0,
        "integration_cache_ttl_sec": 300,
    },
))

Presets:

fast / default: no optional integrations
enriched: grounding + ctx_search
deliberative: grounding + experta + debate
full: all optional integrations

Use integrations=["grounding", "ctx_search"] for exact control.

MCP discovery contract:

get_capabilities returns consultant-specific discovery for all 26 MCP tools: component, domain_ops, shared_ops, discovery_ops, sensitive_ops, integration_presets, grounded_qa, canonical_fields, and reliability_labels.
info returns component="job_consultant", maturity, domain_expert_review verification, R2+ human-review policy for analytical/professional work, the grounded-QA status, and the degradation policy.
Canonical fields are completion_state, warning_card, evidence, request_id, task_id, and run_id.
Reliability labels are exactly verified, qualified-draft, and blocked-escalated.
Deterministic local success is verified; optional integration fallback is qualified-draft with warning_card; no safe consultant output is blocked-escalated.

Consulting decision-support caveat

Use job_consultant for structured analysis, planning, benchmarking, and draft recommendations. Do not treat outputs as professional, legal, financial, HR, procurement, regulatory, or board-level sign-off without qualified human review. Optional grounding, search, debate, Bayesian, and recursive-summary integrations are disabled by default to keep first-run behavior fast and deterministic; when enabled they may add latency, depend on local configuration or external services, and should be checked against source data before being used in client-facing deliverables.

Grounded QA and forbidden claims:

forbidden_claims_scan is a declared failure mode. The available QA harness is grounded_qa.assess_consultant_output, which can be called for grounded review when a consulting output contains unsupported professional claims or will be used in client-facing material. It is optional/on-demand and not automatic per inference. Outputs remain qualified-draft until evidence supports verified; unsupported professional claims should produce warning_card guidance and may become blocked-escalated if required evidence is missing.

Works well with: job_framework, job_business, job_analyst

Public API¶

`JobConsultantBlock(JobAgentBlock)`¶

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

`JobConsultantInput(JobInput)`¶

Input for the Consultant job agent — extended with domain types.

`JobConsultantOutput(JobOutput)`¶

Output from the Consultant job agent — extended with domain data.

`JobConsultantMCPBlock(AIBlock[MCPJobConsultantInput, MCPJobConsultantOutput, dict])`¶

26-op MCP block for the Consultant job agent.

Field	Type	Default
`name`	`str`	'job_consultant_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: MCPJobConsultantInput) -> Result[MCPJobConsultantOutput]`¶

`MCPJobConsultantInput(BaseModel)`¶

Input to JobConsultantMCPBlock - 26-op dispatch.

Field	Type	Default
`op`	`Literal['assess_situation', 'develop_strategy', 'prepare_report', 'deliver_recommendation', 'conduct_review', 'analyze_market', 'evaluate_options', 'assess_impact', 'benchmark_performance', 'audit_process', '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`	''

`MCPJobConsultantOutput(BaseModel)`¶

Output from JobConsultantMCPBlock.

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`	`str`	'qualified-draft'
`warning_card`	`dict[str, Any]`	Field(default_factory=dict)
`evidence`	`dict[str, Any]`	Field(default_factory=dict)

`ConsultantStore(JobStore)`¶

SQLite store for the Consultant job agent.

Constructor:

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

Methods:

`create_engagement(client_name: str = '', project_name: str = '', engagement_type: str = 'advisory', budget: float = 0.0, start_date: str = '', end_date: str = '', data: dict | None = None) -> str`¶

`get_engagement(engagement_id: str) -> dict | None`¶

`list_engagements(status: str = '', limit: int = 50) -> list[dict]`¶

`update_engagement(engagement_id: str, **fields: Any) -> bool`¶

`create_deliverable(engagement_id: str = '', title: str = '', deliverable_type: str = 'report', due_date: str = '', data: dict | None = None) -> str`¶

`get_deliverable(deliverable_id: str) -> dict | None`¶

`list_deliverables(engagement_id: str = '', limit: int = 50) -> list[dict]`¶

`update_deliverable(deliverable_id: str, **fields: Any) -> bool`¶

`create_assessment(engagement_id: str = '', framework: str = '', scope: str = '', findings: dict | None = None, score: float = 0.0) -> str`¶

`get_assessment(assessment_id: str) -> dict | None`¶

`list_assessments(engagement_id: str = '', framework: str = '', limit: int = 50) -> list[dict]`¶

`create_recommendation(engagement_id: str = '', category: str = '', priority: str = 'medium', impact: str = 'medium', effort: str = 'medium', data: dict | None = None) -> str`¶

`list_recommendations(engagement_id: str = '', priority: str = '', limit: int = 50) -> list[dict]`¶

`update_recommendation(rec_id: str, **fields: Any) -> bool`¶

Functions¶

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

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

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

Run grounded four-valued QA over a consultant output.

MCP Tools¶

Operation	Source
`assess_situation`	consultant_mcp
`develop_strategy`	consultant_mcp
`prepare_report`	consultant_mcp
`deliver_recommendation`	consultant_mcp
`conduct_review`	consultant_mcp
`analyze_market`	consultant_mcp
`evaluate_options`	consultant_mcp
`assess_impact`	consultant_mcp
`benchmark_performance`	consultant_mcp
`audit_process`	consultant_mcp
`create_proposal`	consultant_mcp
`review_deliverable`	consultant_mcp
`delegate_task`	consultant_mcp
`report_status`	consultant_mcp
`request_feedback`	consultant_mcp
`store_artifact`	consultant_mcp
`retrieve_artifact`	consultant_mcp
`list_artifacts`	consultant_mcp
`search_artifacts`	consultant_mcp
`archive`	consultant_mcp
`plan_sprint`	consultant_mcp
`track_progress`	consultant_mcp
`reflect_on_outcome`	consultant_mcp
`list_patterns`	consultant_mcp
`get_capabilities`	consultant_mcp
`info`	consultant_mcp

Job Consultant¶

Overview¶

Public API¶

JobConsultantBlock(JobAgentBlock)¶

JobConsultantInput(JobInput)¶

JobConsultantOutput(JobOutput)¶

JobConsultantMCPBlock(AIBlock[MCPJobConsultantInput, MCPJobConsultantOutput, dict])¶

infer(data: MCPJobConsultantInput) -> Result[MCPJobConsultantOutput]¶

MCPJobConsultantInput(BaseModel)¶

MCPJobConsultantOutput(BaseModel)¶

ConsultantStore(JobStore)¶

create_engagement(client_name: str = '', project_name: str = '', engagement_type: str = 'advisory', budget: float = 0.0, start_date: str = '', end_date: str = '', data: dict | None = None) -> str¶

get_engagement(engagement_id: str) -> dict | None¶

list_engagements(status: str = '', limit: int = 50) -> list[dict]¶

update_engagement(engagement_id: str, **fields: Any) -> bool¶

create_deliverable(engagement_id: str = '', title: str = '', deliverable_type: str = 'report', due_date: str = '', data: dict | None = None) -> str¶

get_deliverable(deliverable_id: str) -> dict | None¶

list_deliverables(engagement_id: str = '', limit: int = 50) -> list[dict]¶

update_deliverable(deliverable_id: str, **fields: Any) -> bool¶

create_assessment(engagement_id: str = '', framework: str = '', scope: str = '', findings: dict | None = None, score: float = 0.0) -> str¶

get_assessment(assessment_id: str) -> dict | None¶

list_assessments(engagement_id: str = '', framework: str = '', limit: int = 50) -> list[dict]¶

create_recommendation(engagement_id: str = '', category: str = '', priority: str = 'medium', impact: str = 'medium', effort: str = 'medium', data: dict | None = None) -> str¶

list_recommendations(engagement_id: str = '', priority: str = '', limit: int = 50) -> list[dict]¶

update_recommendation(rec_id: str, **fields: Any) -> bool¶