Job Trades¶

job_trades — G6 Trades job agent.

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

Overview¶

Domain-specialist job agent for skilled tradespeople including electricians, plumbers, carpenters, and HVAC technicians. Diagnoses faults from symptom descriptions, plans repair or installation sequences, sources materials and parts, inspects completed work against standards, documents job records, and analyses failure modes — all within G6's safety-first, audit-trailed JobAgentBlock framework.

When to use:

Diagnosing a reported fault (electrical, plumbing, HVAC) and generating a prioritised investigation sequence
Planning a repair or installation job with materials list, tool requirements, and safety precautions
Documenting completed job records with photos, measurements, and compliance notes
Analysing recurring failure patterns across service history to recommend preventive action

Example:

from mvp.job_trades import JobTradesBlock, JobTradesInput

block = JobTradesBlock()
result = block.infer(JobTradesInput(
    task="Diagnose intermittent RCD tripping on the main switchboard of a commercial kitchen and plan the fault-finding sequence",
    context={"trade": "electrical", "site": "Surry Hills Restaurant", "circuit": "kitchen-main", "frequency": "3x per week"},
))
# result.ok → True; result.value → JobTradesOutput with result, artifacts

Works well with: job_framework, job_labourer, job_machinery_operator

Performance and Enrichment¶

By default, job_trades runs the deterministic trade calculations and safety/HITL checks needed for fast MCP use. This keeps common calls such as diagnose_fault, assess_cost, and benchmark_time responsive for first-run onboarding and pilot workflows.

Expensive enrichment is opt-in. Set parameters.enrichment, parameters.enable_enrichment, context.enrichment, or context.enable_enrichment to true, "standard", "deep", or "full" when you need grounding, web search, debate, Experta, or Bayesian analysis. These modes can add noticeable latency and should be used for deeper review, not every interactive call.

The optional sklearn cost model is also disabled by default. To run it during cost assessment, enable enrichment and set parameters.enable_ml_cost_model=True.

Public API¶

`JobTradesInput(JobInput)`¶

Input for the Trades job agent.

`JobTradesOutput(JobOutput)`¶

`JobTradesBlock(JobAgentBlock)`¶

G6 Trades job agent — Tier 1 block with MCP delegation and two safety floors.

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

`MCPJobTradesInput(BaseModel)`¶

Input to JobTradesMCPBlock — 26-op dispatch.

Field	Type	Default
`op`	`Literal['diagnose_fault', 'plan_repair', 'source_materials', 'inspect_work', 'document_job', 'analyze_failure', 'evaluate_options', 'assess_cost', 'benchmark_time', 'audit_workmanship', '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`	''

`MCPJobTradesOutput(BaseModel)`¶

Output from JobTradesMCPBlock.

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
`human_review_required`	`bool`	False

`TradesStore(JobStore)`¶

SQLite store for the Trades job agent.

Constructor:

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

Methods:

`create_job(trade: str, description: str = '', location: str = '', client: str = '', priority: str = 'normal', estimated_hours: float = 0, estimated_cost: float = 0, data: dict | None = None) -> str`¶

`get_job(job_id: str) -> dict | None`¶

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

`update_job(job_id: str, **fields: Any) -> bool`¶

`create_permit(job_id: str = '', permit_type: str = '', jurisdiction: str = '', notes: str = '', data: dict | None = None) -> str`¶

`get_permit(permit_id: str) -> dict | None`¶

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

`update_permit(permit_id: str, **fields: Any) -> bool`¶

`create_inspection(job_id: str = '', trade: str = '', inspection_type: str = '', inspector: str = '', findings: list | None = None, checklist: dict | None = None, pass_fail: str = 'pending', data: dict | None = None) -> str`¶

`list_inspections(job_id: str = '', trade: str = '', pass_fail: str = '', limit: int = 50) -> list[dict]`¶

`create_material(job_id: str = '', name: str = '', category: str = '', quantity: float = 0, unit: str = 'ea', unit_cost: float = 0, supplier: str = '', data: dict | None = None) -> str`¶

`list_materials(job_id: str = '', category: str = '', status: str = '', limit: int = 100) -> list[dict]`¶

`get_material_summary(job_id: str) -> dict`¶

`create_apprentice(name: str, trade: str, level: int = 1, hours_required: float = 8000, journeyman_id: str = '', data: dict | None = None) -> str`¶

`log_apprentice_hours(apprentice_id: str, hours: float, notes: str = '') -> bool`¶

`get_apprentice(apprentice_id: str) -> dict | None`¶

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

`create_work_item(category: str, title: str, data: dict | None = None) -> str`¶

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

`JobTradesMCPBlock(AIBlock[MCPJobTradesInput, MCPJobTradesOutput, dict])`¶

26-op MCP block for the Trades job agent.

Field	Type	Default
`name`	`str`	'job_trades_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: MCPJobTradesInput) -> Result[MCPJobTradesOutput]`¶

Functions¶

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

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

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

Run grounded four-valued QA over a trades output.

MCP Tools¶

Operation	Source
`diagnose_fault`	trades_mcp
`plan_repair`	trades_mcp
`source_materials`	trades_mcp
`inspect_work`	trades_mcp
`document_job`	trades_mcp
`analyze_failure`	trades_mcp
`evaluate_options`	trades_mcp
`assess_cost`	trades_mcp
`benchmark_time`	trades_mcp
`audit_workmanship`	trades_mcp
`create_proposal`	trades_mcp
`review_deliverable`	trades_mcp
`delegate_task`	trades_mcp
`report_status`	trades_mcp
`request_feedback`	trades_mcp
`store_artifact`	trades_mcp
`retrieve_artifact`	trades_mcp
`list_artifacts`	trades_mcp
`search_artifacts`	trades_mcp
`archive`	trades_mcp
`plan_sprint`	trades_mcp
`track_progress`	trades_mcp
`reflect_on_outcome`	trades_mcp
`get_capabilities`	trades_mcp
`info`	trades_mcp
`list_patterns`	trades_mcp

Job Trades¶

Overview¶

Performance and Enrichment¶

Public API¶

JobTradesInput(JobInput)¶

JobTradesOutput(JobOutput)¶

JobTradesBlock(JobAgentBlock)¶

MCPJobTradesInput(BaseModel)¶

MCPJobTradesOutput(BaseModel)¶

TradesStore(JobStore)¶

create_job(trade: str, description: str = '', location: str = '', client: str = '', priority: str = 'normal', estimated_hours: float = 0, estimated_cost: float = 0, data: dict | None = None) -> str¶

get_job(job_id: str) -> dict | None¶

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

update_job(job_id: str, **fields: Any) -> bool¶

create_permit(job_id: str = '', permit_type: str = '', jurisdiction: str = '', notes: str = '', data: dict | None = None) -> str¶

get_permit(permit_id: str) -> dict | None¶

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

update_permit(permit_id: str, **fields: Any) -> bool¶

create_inspection(job_id: str = '', trade: str = '', inspection_type: str = '', inspector: str = '', findings: list | None = None, checklist: dict | None = None, pass_fail: str = 'pending', data: dict | None = None) -> str¶

list_inspections(job_id: str = '', trade: str = '', pass_fail: str = '', limit: int = 50) -> list[dict]¶

create_material(job_id: str = '', name: str = '', category: str = '', quantity: float = 0, unit: str = 'ea', unit_cost: float = 0, supplier: str = '', data: dict | None = None) -> str¶

list_materials(job_id: str = '', category: str = '', status: str = '', limit: int = 100) -> list[dict]¶

get_material_summary(job_id: str) -> dict¶

create_apprentice(name: str, trade: str, level: int = 1, hours_required: float = 8000, journeyman_id: str = '', data: dict | None = None) -> str¶

log_apprentice_hours(apprentice_id: str, hours: float, notes: str = '') -> bool¶

get_apprentice(apprentice_id: str) -> dict | None¶

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

create_work_item(category: str, title: str, data: dict | None = None) -> str¶

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

JobTradesMCPBlock(AIBlock[MCPJobTradesInput, MCPJobTradesOutput, dict])¶

infer(data: MCPJobTradesInput) -> Result[MCPJobTradesOutput]¶