Skip to content

Job Labourer

job_labourer — G6 Labourer job agent.

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

Overview

Domain-specialist job agent supporting labourers and frontline workers. Logs daily tasks and time records, follows and validates safety procedures, reports hazards, tracks hours against project allocations, manages equipment requests, and analyses workload distribution — all within G6's safety-first, audit-trailed JobAgentBlock framework.

When to use:

  • Logging task completion records and time entries for payroll or project billing
  • Reporting a safety hazard with structured evidence capture for incident management
  • Requesting equipment or consumables against a project budget
  • Reviewing workload distribution across a crew for equitable allocation

Example:

from mvp.job_labourer import JobLabourerBlock, JobLabourerInput

block = JobLabourerBlock()
result = block.infer(JobLabourerInput(
    task="Log today's site tasks, hours worked, and report the unsecured scaffolding hazard at Grid E4",
    context={"site": "Parramatta-CBD-Stage2", "crew_id": "C-14", "date": "2026-04-07"},
))
# result.ok → True; result.value → JobLabourerOutput with result, artifacts

Works well with: job_framework, job_construction, job_trades

Public API

JobLabourerBlock(JobAgentBlock)

G6 Labourer job agent -- Tier 1 block with MCP delegation and safety gate.

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

Methods:

infer(data)

JobLabourerInput(JobInput)

Input for the Labourer job agent.

JobLabourerOutput(JobOutput)

JobLabourerMCPBlock(AIBlock[MCPJobLabourerInput, MCPJobLabourerOutput, dict])

26-op MCP block for the Labourer job agent.

Field Type Default
name str 'job_labourer_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: MCPJobLabourerInput) -> Result[MCPJobLabourerOutput]

MCPJobLabourerInput(BaseModel)

Input to JobLabourerMCPBlock -- 26-op dispatch.

Field Type Default
op Literal['log_task', 'follow_procedure', 'report_hazard', 'track_hours', 'request_equipment', 'analyze_workload', 'evaluate_safety', 'assess_conditions', 'benchmark_output', 'audit_compliance', 'list_patterns', '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'] required
task str ''
context dict[str, Any] Field(default_factory=dict)
parameters dict[str, Any] Field(default_factory=dict)
artifact_id str ''
query str ''

MCPJobLabourerOutput(BaseModel)

Output from JobLabourerMCPBlock.

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)
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)
request_id str ''
task_id str ''
run_id str ''
degraded bool False
degradation_reason str \| None None

LabourerStore(JobStore)

SQLite store for the Labourer job agent.

Constructor:

Parameter Type Default
db_path str ':memory:'

Methods:

db_path() -> str

save_task(task_data: dict) -> str

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

save_safety_report(report: dict) -> str

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

save_timesheet(ts: dict) -> str

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

save_equipment_request(req: dict) -> str

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

save_incident(inc: dict) -> str

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

save_training_record(rec: dict) -> str

get_training_records(worker_id: str = '', limit: int = 50) -> list[dict]

save_quality_inspection(insp: dict) -> str

get_quality_inspections(task_id: str = '', limit: int = 50) -> list[dict]

save_work_entry(category: str, title: str, data: dict) -> str

get_work_entries(category: str = '', limit: int = 50) -> list[dict]

Functions

assemble_review_text(output: Any) -> str

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

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

Run grounded four-valued QA over a labourer output.

MCP Tools

Operation Source
log_task labourer_mcp
follow_procedure labourer_mcp
report_hazard labourer_mcp
track_hours labourer_mcp
request_equipment labourer_mcp
analyze_workload labourer_mcp
evaluate_safety labourer_mcp
assess_conditions labourer_mcp
benchmark_output labourer_mcp
audit_compliance labourer_mcp
list_patterns labourer_mcp
create_proposal labourer_mcp
review_deliverable labourer_mcp
delegate_task labourer_mcp
report_status labourer_mcp
request_feedback labourer_mcp
store_artifact labourer_mcp
retrieve_artifact labourer_mcp
list_artifacts labourer_mcp
search_artifacts labourer_mcp
archive labourer_mcp
plan_sprint labourer_mcp
track_progress labourer_mcp
reflect_on_outcome labourer_mcp
get_capabilities labourer_mcp
info labourer_mcp