Skip to content

Job Construction¶

job_construction — G6 Construction job agent.

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

Overview¶

Domain-specialist job agent for construction project management. Produces cost estimates, phase schedules, site inspection reports, materials management plans, progress reports, and cost analysis — covering the full construction project lifecycle within G6's safety-bounded, audit-trailed JobAgentBlock framework.

Pilot-ready, not compliance-grade

job_construction is suitable for MVP and design-partner pilot workflows where a qualified person reviews the result. It is not a regulated construction, safety, engineering, certification, or compliance approval system. Safety-critical operations such as inspect_site, audit_safety, and assess_risk require professional or research context and emit construction disclaimers, but the current gate does not verify licences, credentials, permits, site conditions, or jurisdiction-specific code compliance. human_review_required=True is advisory, not a hard execution block. High-stakes outputs use completion_state: qualified-draft, review_status: pending, and warning code construction_high_stakes_review_required until reviewed. Do not market or use this component as autonomous construction safety approval or enterprise compliance automation.

Canonical envelope:

Outputs preserve completion_state, warning_card, evidence, request_id, task_id, and run_id
Legal completion_state values are exactly verified, qualified-draft, and blocked-escalated
Deterministic local success is verified; degraded-but-useful output is qualified-draft; refused, not-found, or failed output is blocked-escalated
get_capabilities exposes tier1_routable_ops, mcp_direct_ops, high_stakes_ops, optional backend availability, store health, and web policy

Network policy:

DuckDuckGo-backed generic search is disabled by default and only runs when use_web_search=True
Perplexity-backed grounding remains disabled by default and only runs when use_perplexity=True

When to use:

Generating detailed cost estimates and bill-of-quantities for a new construction project
Building phased Gantt-style schedules from scope and resource inputs
Producing structured site-inspection reports with defect categorisation and remediation steps
Monitoring materials consumption against budget and flagging overruns early

Example:

from mvp.job_construction import JobConstructionBlock, JobConstructionInput

block = JobConstructionBlock()
result = block.infer(JobConstructionInput(
    task="Estimate total cost and phase schedule for a 20-unit residential development in Brisbane",
    context={"land_area_m2": 4000, "storeys": 3, "start_date": "2026-07-01"},
))
# result.ok → True; result.value → JobConstructionOutput with result, artifacts

Works well with: job_framework, job_engineer, job_manager

Public API¶

`JobConstructionBlock(JobAgentBlock)`¶

G6 Construction job agent - Tier 1 block with MCP delegation.

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

Methods:

`infer(data) -> Result[JobConstructionOutput]`¶

`JobConstructionInput(JobInput)`¶

Input for the Construction job agent.

`JobConstructionOutput(JobOutput)`¶

`JobConstructionMCPBlock(AIBlock[MCPJobConstructionInput, MCPJobConstructionOutput, dict])`¶

26-op MCP block for the Construction job agent.

Field	Type	Default
`name`	`str`	'job_construction_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: MCPJobConstructionInput) -> Result[MCPJobConstructionOutput]`¶

`MCPJobConstructionInput(BaseModel)`¶

Input to JobConstructionMCPBlock - 26-op dispatch.

Field	Type	Default
`op`	`Literal['estimate_project', 'schedule_phases', 'inspect_site', 'manage_materials', 'report_progress', 'analyze_costs', 'evaluate_timeline', 'assess_risk', 'benchmark_productivity', 'audit_safety', '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`	''
`request_id`	`str`	''
`task_id`	`str`	''
`run_id`	`str`	''

`MCPJobConstructionOutput(BaseModel)`¶

Output from JobConstructionMCPBlock.

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`	`CompletionState`	'qualified-draft'
`warning_card`	`dict[str, Any]`	Field(default_factory=dict)
`evidence`	`dict[str, Any]`	Field(default_factory=dict)
`human_review_required`	`bool`	False
`requires_human_review`	`bool`	False
`review_status`	`str`	''
`request_id`	`str`	''
`task_id`	`str`	''
`run_id`	`str`	''

`ConstructionStore(JobStore)`¶

SQLite store for the Construction job agent.

Constructor:

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

Methods:

`save_project(project_id: str = '', name: str = '', project_type: str = '', client: str = '', location: str = '', contract_value: float = 0, start_date: str = '', completion_date: str = '', status: str = 'planning', data: dict | None = None) -> str`¶

Insert or replace a project record. Returns the project_id.

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

Fetch a project by ID.

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

List projects, optionally filtering by status.

`save_estimate(project_id: str = '', category: str = '', description: str = '', quantity: float = 0, unit: str = 'ea', unit_cost: float = 0, total: float = 0, markup: float = 0, data: dict | None = None) -> str`¶

Insert an estimate line item. Returns the estimate_id.

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

Get all estimate line items for a project.

`get_estimate_total(project_id: str) -> float`¶

Sum of all estimate totals for a project.

`save_schedule(project_id: str = '', task_name: str = '', wbs_code: str = '', duration_days: int = 0, start_date: str = '', end_date: str = '', predecessors: list[str] | None = None, resources: list[str] | None = None, status: str = 'planned') -> str`¶

Insert a schedule activity. Returns the schedule_id.

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

Get all schedule activities for a project.

`save_inspection(project_id: str = '', inspection_type: str = '', inspector: str = '', findings: dict | None = None, pass_fail: str = 'pending', deficiencies: list[str] | None = None, inspection_date: str = '') -> str`¶

Insert an inspection record. Returns the inspection_id.

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

Get all inspections for a project.

`save_change_order(project_id: str = '', description: str = '', cost_impact: float = 0, schedule_impact_days: int = 0, justification: str = '', approval_status: str = 'pending', data: dict | None = None) -> str`¶

Insert a change order. Returns the co_id.

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

Get all change orders for a project.

`get_co_total(project_id: str) -> float`¶

Sum of all change order cost impacts for a project.

`save_equipment(project_id: str = '', name: str = '', equipment_type: str = '', daily_rate: float = 0, status: str = 'available', hours_used: float = 0, last_service: str = '', data: dict | None = None) -> str`¶

Insert an equipment record. Returns the equipment_id.

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

Get all equipment for a project.

`save_subcontractor(company_name: str = '', trade: str = '', license_number: str = '', bonding_capacity: float = 0, rating: float = 0, status: str = 'active', data: dict | None = None) -> str`¶

Insert a subcontractor record. Returns the sub_id.

`get_subcontractors(trade: str = '') -> list[dict]`¶

Get subcontractors, optionally filtered by trade.

Functions¶

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

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

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

Run grounded four-valued QA over a construction output.

MCP Tools¶

Operation	Source
`verified`	construction_mcp
`qualified-draft`	construction_mcp
`blocked-escalated`	construction_mcp
`estimate_project`	construction_mcp
`schedule_phases`	construction_mcp
`inspect_site`	construction_mcp
`manage_materials`	construction_mcp
`report_progress`	construction_mcp
`analyze_costs`	construction_mcp
`evaluate_timeline`	construction_mcp
`assess_risk`	construction_mcp
`benchmark_productivity`	construction_mcp
`audit_safety`	construction_mcp
`create_proposal`	construction_mcp
`review_deliverable`	construction_mcp
`delegate_task`	construction_mcp
`report_status`	construction_mcp
`request_feedback`	construction_mcp
`store_artifact`	construction_mcp
`retrieve_artifact`	construction_mcp
`list_artifacts`	construction_mcp
`search_artifacts`	construction_mcp
`archive`	construction_mcp
`plan_sprint`	construction_mcp
`track_progress`	construction_mcp
`reflect_on_outcome`	construction_mcp
`list_patterns`	construction_mcp
`get_capabilities`	construction_mcp
`info`	construction_mcp