Job Machinery Operator¶
job_machinery_operator — G6 MachineryOperator job agent.
Cluster: Job Agents | Type: component | MCP Tools: 26
Overview¶
Domain-specialist job agent for heavy machinery and plant operators. Logs equipment operations, schedules and tracks maintenance, reports faults with structured diagnostics, calibrates instruments, checks safety pre-start checklists, and analyses downtime patterns — all within G6's safety-first, audit-trailed JobAgentBlock framework.
When to use:
- Logging equipment operating hours and triggering preventive maintenance alerts
- Filing structured fault reports with symptom description, probable cause, and urgency rating
- Running pre-start safety checklists and capturing evidence for compliance records
- Analysing historical downtime to identify recurring failure modes and schedule interventions
Example:
from mvp.job_machinery_operator import JobMachineryOperatorBlock, JobMachineryOperatorInput
block = JobMachineryOperatorBlock()
result = block.infer(JobMachineryOperatorInput(
task="Log the excavator CAT 390F pre-start check, report abnormal hydraulic pressure reading, and flag for urgent maintenance",
context={"machine_id": "EX-390F-07", "site": "Kingsford-Smith-Runway-Extension", "shift": "day"},
))
# result.ok → True; result.value → JobMachineryOperatorOutput with result, artifacts
Works well with: job_framework, job_trades, job_manufacturing
Public API¶
JobMachineryOperatorBlock(JobAgentBlock)¶
G6 MachineryOperator job agent — Tier 1 block with MCP delegation and safety gate.
| Field | Type | Default |
|---|---|---|
name | str | 'job_machinery_operator' |
sector | SectorClassification | field(default_factory=lambda: _SECTOR) |
toolkit | ToolkitSpec \| None | field(default_factory=lambda: JOB_TOOLKITS.get('machinery_operator')) |
mcp_module | str | 'mvp.job_machinery_operator.machinery_operator_mcp.server' |
capabilities | ClassVar[set[type]] | {Extensible, HumanLearnable, Collaborative, ProblemSolvable, KnowledgeGrounded, Memorable, AgentCommunicable, Actuatable} |
Methods:
infer(data: Any) -> Result[JobMachineryOperatorOutput]¶
JobMachineryOperatorInput(JobInput)¶
Input for the MachineryOperator job agent.
JobMachineryOperatorOutput(JobOutput)¶
JobMachineryOperatorMCPBlock(AIBlock[MCPJobMachineryOperatorInput, MCPJobMachineryOperatorOutput, dict])¶
26-op MCP block for the MachineryOperator job agent.
| Field | Type | Default |
|---|---|---|
name | str | 'job_machinery_operator_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: MCPJobMachineryOperatorInput) -> Result[MCPJobMachineryOperatorOutput]¶
MCPJobMachineryOperatorInput(BaseModel)¶
Input to JobMachineryOperatorMCPBlock — 26-op dispatch.
| Field | Type | Default |
|---|---|---|
op | Literal['operate_equipment', 'log_maintenance', 'report_fault', 'calibrate_machine', 'check_safety', 'analyze_downtime', 'evaluate_performance', 'assess_wear', 'benchmark_efficiency', 'audit_logs', '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 | '' |
MCPJobMachineryOperatorOutput(BaseModel)¶
Output from JobMachineryOperatorMCPBlock.
| 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) |
human_review_required | bool | False |
MachineryOperatorStore(JobStore)¶
SQLite store for the MachineryOperator job agent.
Constructor:
| Parameter | Type | Default |
|---|---|---|
db_path | str | ':memory:' |
Methods:
save_machine(machine_id: str = '', name: str = '', machine_type: str = '', manufacturer: str = '', model: str = '', serial_number: str = '', year: int = 0, rated_power_kw: float = 0, fuel_type: str = 'diesel', status: str = 'available', location: str = '', hourometer: float = 0, purchase_cost: float = 0, salvage_value: float = 0, useful_life_hrs: float = 10000, data: dict | None = None) -> str¶
get_machine(machine_id: str) -> dict | None¶
list_machines(machine_type: str = '', status: str = '') -> list[dict]¶
save_operation_log(machine_id: str = '', operator_id: str = '', operation_type: str = '', start_time: str = '', end_time: str = '', hours_operated: float = 0, fuel_consumed: float = 0, work_output: float = 0, output_unit: str = '', notes: str = '', data: dict | None = None) -> str¶
get_operation_logs(machine_id: str) -> list[dict]¶
get_total_hours(machine_id: str) -> float¶
save_maintenance(machine_id: str = '', maint_type: str = 'preventive', description: str = '', cost: float = 0, downtime_hours: float = 0, parts_replaced: list[str] | None = None, technician: str = '', scheduled_date: str = '', completed_date: str = '', status: str = 'pending', data: dict | None = None) -> str¶
get_maintenance_records(machine_id: str) -> list[dict]¶
get_maintenance_cost_total(machine_id: str) -> float¶
get_total_downtime(machine_id: str) -> float¶
save_certification(operator_id: str = '', operator_name: str = '', cert_type: str = '', machine_types: list[str] | None = None, issue_date: str = '', expiry_date: str = '', issuing_body: str = '', experience_hrs: float = 0, skill_level: str = 'beginner', status: str = 'active', data: dict | None = None) -> str¶
get_certifications(operator_id: str) -> list[dict]¶
get_expiring_certs(days_ahead: int = 90) -> list[dict]¶
save_inspection(machine_id: str = '', inspector_id: str = '', inspection_type: str = 'pre_start', checklist_items: list[str] | None = None, passed_items: int = 0, failed_items: int = 0, overall_status: str = 'pending', notes: str = '', inspection_date: str = '', data: dict | None = None) -> str¶
get_inspections(machine_id: str) -> list[dict]¶
save_cost_record(machine_id: str = '', cost_type: str = '', amount: float = 0, period: str = '', description: str = '', data: dict | None = None) -> str¶
get_cost_records(machine_id: str, cost_type: str = '') -> list[dict]¶
get_total_cost(machine_id: str, cost_type: str = '') -> float¶
save_production_log(machine_id: str = '', operator_id: str = '', shift_date: str = '', shift_type: str = 'day', planned_hours: float = 0, actual_hours: float = 0, downtime_hours: float = 0, output_quantity: float = 0, output_unit: str = '', fuel_consumed: float = 0, oee_score: float = 0, data: dict | None = None) -> str¶
get_production_logs(machine_id: str, limit: int = 50) -> list[dict]¶
get_avg_oee(machine_id: str) -> float¶
Functions¶
assemble_review_text(output: Any) -> str¶
Collect the reviewable free text from a JobMachineryOperatorOutput (duck-typed).
assess_machinery_operator_output(output: Any, qa_block: Any | None = None, generate: Any | None = None) -> GroundedRunResult¶
Run grounded four-valued QA over a machinery-operator output.
MCP Tools¶
| Operation | Source |
|---|---|
operate_equipment | machinery_operator_mcp |
log_maintenance | machinery_operator_mcp |
report_fault | machinery_operator_mcp |
calibrate_machine | machinery_operator_mcp |
check_safety | machinery_operator_mcp |
analyze_downtime | machinery_operator_mcp |
evaluate_performance | machinery_operator_mcp |
assess_wear | machinery_operator_mcp |
benchmark_efficiency | machinery_operator_mcp |
audit_logs | machinery_operator_mcp |
create_proposal | machinery_operator_mcp |
review_deliverable | machinery_operator_mcp |
delegate_task | machinery_operator_mcp |
report_status | machinery_operator_mcp |
request_feedback | machinery_operator_mcp |
store_artifact | machinery_operator_mcp |
retrieve_artifact | machinery_operator_mcp |
list_artifacts | machinery_operator_mcp |
search_artifacts | machinery_operator_mcp |
archive | machinery_operator_mcp |
plan_sprint | machinery_operator_mcp |
track_progress | machinery_operator_mcp |
reflect_on_outcome | machinery_operator_mcp |
get_capabilities | machinery_operator_mcp |
info | machinery_operator_mcp |
list_patterns | machinery_operator_mcp |