Job Energy¶

job_energy — G6 Energy job agent.

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

Overview¶

Domain-specialist job agent for energy sector professionals. Monitors grid operations, optimises power distribution, forecasts demand curves, inspects facilities, reports generation output, and analyses consumption patterns — providing a structured energy-management workflow within G6's safety-bounded, audit-trailed JobAgentBlock framework.

When to use:

Monitoring real-time grid telemetry and detecting anomalies or fault conditions
Optimising dispatch schedules to balance renewable intermittency against baseload demand
Producing regulatory compliance reports on generation output and emissions
Forecasting short- and medium-term energy demand to inform procurement decisions

Production caveats

job_energy is an advisory analysis component, not operational authority for grid, electrical, process-safety, trading, or energy-facility actions. Outputs must be reviewed by qualified operators, licensed engineers, or other responsible professionals before use in real-world operations, compliance filings, switching decisions, dispatch changes, facility work, or safety-critical remediation.

High-stakes operations such as monitor_grid, optimize_distribution, forecast_demand, inspect_facility, and audit_emissions require an explicit professional context or context={"mode": "research_only"}. Without that context, the top-level job agent returns completion_state="blocked-escalated" with confidence=0.0, a warning_card, and structured evidence instead of executing the operation. Deterministic fallback remains useful but is surfaced as completion_state="qualified-draft" rather than overclaimed as verified.

Live web search and live LLM-backed debate are opt-in (G6_JOB_LIVE_SEARCH=1 and G6_JOB_LIVE_LLM=1). The default mode is local/deterministic so first-time MCP installs, CI, and pilot demos do not depend on external network calls or paid model credentials.

Example:

from mvp.job_energy import JobEnergyBlock, JobEnergyInput

block = JobEnergyBlock()
result = block.infer(JobEnergyInput(
    task="Forecast electricity demand for the Sydney metro grid over the next 7 days and recommend dispatch schedule",
    context={"region": "NSW", "renewable_capacity_mw": 2400, "horizon_days": 7},
))
# result.ok → True; result.value → JobEnergyOutput with result, artifacts

Works well with: job_framework, job_engineer, job_scientist

Public API¶

`JobEnergyBlock(JobAgentBlock)`¶

G6 Energy job agent — Tier 1 block with MCP delegation.

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

Methods:

`infer(data: JobEnergyInput) -> Result[JobEnergyOutput]`¶

`JobEnergyInput(JobInput)`¶

Input for the Energy job agent.

`JobEnergyOutput(JobOutput)`¶

`JobEnergyMCPBlock(AIBlock[MCPJobEnergyInput, MCPJobEnergyOutput, dict])`¶

26-op MCP block for the Energy job agent.

Field	Type	Default
`name`	`str`	'job_energy_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: MCPJobEnergyInput) -> Result[MCPJobEnergyOutput]`¶

`MCPJobEnergyInput(BaseModel)`¶

Input to JobEnergyMCPBlock -- 26-op dispatch.

Field	Type	Default
`op`	`Literal['monitor_grid', 'optimize_distribution', 'forecast_demand', 'inspect_facility', 'report_output', 'analyze_consumption', 'evaluate_efficiency', 'assess_capacity', 'benchmark_generation', 'audit_emissions', '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`	''

`MCPJobEnergyOutput(BaseModel)`¶

Output from JobEnergyMCPBlock.

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

`EnergyStore(JobStore)`¶

SQLite store for the Energy job agent.

Constructor:

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

Methods:

`create_asset(name: str, fuel_type: str = '', technology: str = '', nameplate_mw: float = 0, location: str = '', commissioned: str = '', data: dict | None = None) -> str`¶

`get_asset(asset_id: str) -> dict | None`¶

`list_assets(fuel_type: str = '', status: str = '') -> list[dict]`¶

`update_asset_status(asset_id: str, status: str) -> bool`¶

`record_production(asset_id: str, period: str, generation_mwh: float = 0, capacity_factor: float = 0, heat_rate: float = 0, emissions_tco2: float = 0, data: dict | None = None) -> str`¶

`get_production(asset_id: str = '', period: str = '', limit: int = 100) -> list[dict]`¶

`get_total_generation(period: str = '') -> float`¶

Sum of generation_mwh for a given period (or all).

`record_grid_metrics(region: str, period: str, peak_demand_mw: float = 0, avg_demand_mw: float = 0, reserve_margin: float = 0, frequency_hz: float = 60.0, saidi_minutes: float = 0, data: dict | None = None) -> str`¶

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

`get_latest_grid_metric(region: str = '') -> dict | None`¶

Get the most recent grid metric record.

`create_filing(filing_type: str, authority: str = '', due_date: str = '', data: dict | None = None) -> str`¶

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

`update_filing_status(filing_id: str, status: str, filed_date: str = '') -> bool`¶

`create_position(commodity: str, direction: str = 'long', quantity: float = 0, entry_price: float = 0, current_price: float = 0, contract_type: str = 'spot', expiry_date: str = '', data: dict | None = None) -> str`¶

`get_positions(commodity: str = '', limit: int = 100) -> list[dict]`¶

`update_position_price(position_id: str, current_price: float) -> bool`¶

`close_position(position_id: str, close_price: float) -> bool`¶

Mark a position as closed with final price.

`record_consumption(facility_id: str, period: str, consumption_kwh: float = 0, peak_demand_kw: float = 0, energy_cost: float = 0, demand_cost: float = 0, facility_type: str = '', data: dict | None = None) -> str`¶

`get_consumption(facility_id: str = '', period: str = '', limit: int = 100) -> list[dict]`¶

`create_storage_system(name: str, technology: str = 'li_ion', capacity_kwh: float = 0, power_kw: float = 0, location: str = '', data: dict | None = None) -> str`¶

`get_storage_system(system_id: str) -> dict | None`¶

`list_storage_systems(technology: str = '', status: str = '') -> list[dict]`¶

`update_storage_health(system_id: str, soh_pct: float, cycles_completed: int) -> bool`¶

`record_emissions(period: str, scope: str = 'scope1', emissions_tco2: float = 0, source: str = '', carbon_intensity: float = 0, data: dict | None = None) -> str`¶

`get_emissions(period: str = '', scope: str = '', limit: int = 100) -> list[dict]`¶

`get_total_emissions(period: str = '', scope: str = '') -> float`¶

Sum emissions for a period/scope.

Functions¶

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

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

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

Run grounded four-valued QA over an energy output.

MCP Tools¶

Operation	Source
`monitor_grid`	energy_mcp
`optimize_distribution`	energy_mcp
`forecast_demand`	energy_mcp
`inspect_facility`	energy_mcp
`report_output`	energy_mcp
`analyze_consumption`	energy_mcp
`evaluate_efficiency`	energy_mcp
`assess_capacity`	energy_mcp
`benchmark_generation`	energy_mcp
`audit_emissions`	energy_mcp
`create_proposal`	energy_mcp
`review_deliverable`	energy_mcp
`delegate_task`	energy_mcp
`report_status`	energy_mcp
`request_feedback`	energy_mcp
`store_artifact`	energy_mcp
`retrieve_artifact`	energy_mcp
`list_artifacts`	energy_mcp
`search_artifacts`	energy_mcp
`archive`	energy_mcp
`plan_sprint`	energy_mcp
`track_progress`	energy_mcp
`reflect_on_outcome`	energy_mcp
`list_patterns`	energy_mcp
`get_capabilities`	energy_mcp
`info`	energy_mcp

Job Energy¶

Overview¶

Public API¶

JobEnergyBlock(JobAgentBlock)¶

infer(data: JobEnergyInput) -> Result[JobEnergyOutput]¶

JobEnergyInput(JobInput)¶

JobEnergyOutput(JobOutput)¶

JobEnergyMCPBlock(AIBlock[MCPJobEnergyInput, MCPJobEnergyOutput, dict])¶

infer(data: MCPJobEnergyInput) -> Result[MCPJobEnergyOutput]¶

MCPJobEnergyInput(BaseModel)¶

MCPJobEnergyOutput(BaseModel)¶

EnergyStore(JobStore)¶

create_asset(name: str, fuel_type: str = '', technology: str = '', nameplate_mw: float = 0, location: str = '', commissioned: str = '', data: dict | None = None) -> str¶

get_asset(asset_id: str) -> dict | None¶

list_assets(fuel_type: str = '', status: str = '') -> list[dict]¶

update_asset_status(asset_id: str, status: str) -> bool¶

record_production(asset_id: str, period: str, generation_mwh: float = 0, capacity_factor: float = 0, heat_rate: float = 0, emissions_tco2: float = 0, data: dict | None = None) -> str¶

get_production(asset_id: str = '', period: str = '', limit: int = 100) -> list[dict]¶

get_total_generation(period: str = '') -> float¶

record_grid_metrics(region: str, period: str, peak_demand_mw: float = 0, avg_demand_mw: float = 0, reserve_margin: float = 0, frequency_hz: float = 60.0, saidi_minutes: float = 0, data: dict | None = None) -> str¶

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

get_latest_grid_metric(region: str = '') -> dict | None¶

create_filing(filing_type: str, authority: str = '', due_date: str = '', data: dict | None = None) -> str¶

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

update_filing_status(filing_id: str, status: str, filed_date: str = '') -> bool¶

create_position(commodity: str, direction: str = 'long', quantity: float = 0, entry_price: float = 0, current_price: float = 0, contract_type: str = 'spot', expiry_date: str = '', data: dict | None = None) -> str¶

get_positions(commodity: str = '', limit: int = 100) -> list[dict]¶

update_position_price(position_id: str, current_price: float) -> bool¶

close_position(position_id: str, close_price: float) -> bool¶

record_consumption(facility_id: str, period: str, consumption_kwh: float = 0, peak_demand_kw: float = 0, energy_cost: float = 0, demand_cost: float = 0, facility_type: str = '', data: dict | None = None) -> str¶

get_consumption(facility_id: str = '', period: str = '', limit: int = 100) -> list[dict]¶

create_storage_system(name: str, technology: str = 'li_ion', capacity_kwh: float = 0, power_kw: float = 0, location: str = '', data: dict | None = None) -> str¶

get_storage_system(system_id: str) -> dict | None¶

list_storage_systems(technology: str = '', status: str = '') -> list[dict]¶

update_storage_health(system_id: str, soh_pct: float, cycles_completed: int) -> bool¶

record_emissions(period: str, scope: str = 'scope1', emissions_tco2: float = 0, source: str = '', carbon_intensity: float = 0, data: dict | None = None) -> str¶

get_emissions(period: str = '', scope: str = '', limit: int = 100) -> list[dict]¶

get_total_emissions(period: str = '', scope: str = '') -> float¶

Functions¶

assemble_review_text(output: Any) -> str¶

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

MCP Tools¶

`JobEnergyBlock(JobAgentBlock)`¶

`infer(data: JobEnergyInput) -> Result[JobEnergyOutput]`¶

`JobEnergyInput(JobInput)`¶

`JobEnergyOutput(JobOutput)`¶

`JobEnergyMCPBlock(AIBlock[MCPJobEnergyInput, MCPJobEnergyOutput, dict])`¶

`infer(data: MCPJobEnergyInput) -> Result[MCPJobEnergyOutput]`¶

`MCPJobEnergyInput(BaseModel)`¶

`MCPJobEnergyOutput(BaseModel)`¶

`EnergyStore(JobStore)`¶

`create_asset(name: str, fuel_type: str = '', technology: str = '', nameplate_mw: float = 0, location: str = '', commissioned: str = '', data: dict | None = None) -> str`¶

`get_asset(asset_id: str) -> dict | None`¶

`list_assets(fuel_type: str = '', status: str = '') -> list[dict]`¶

`update_asset_status(asset_id: str, status: str) -> bool`¶

`record_production(asset_id: str, period: str, generation_mwh: float = 0, capacity_factor: float = 0, heat_rate: float = 0, emissions_tco2: float = 0, data: dict | None = None) -> str`¶

`get_production(asset_id: str = '', period: str = '', limit: int = 100) -> list[dict]`¶

`get_total_generation(period: str = '') -> float`¶

`record_grid_metrics(region: str, period: str, peak_demand_mw: float = 0, avg_demand_mw: float = 0, reserve_margin: float = 0, frequency_hz: float = 60.0, saidi_minutes: float = 0, data: dict | None = None) -> str`¶

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

`get_latest_grid_metric(region: str = '') -> dict | None`¶

`create_filing(filing_type: str, authority: str = '', due_date: str = '', data: dict | None = None) -> str`¶

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

`update_filing_status(filing_id: str, status: str, filed_date: str = '') -> bool`¶

`create_position(commodity: str, direction: str = 'long', quantity: float = 0, entry_price: float = 0, current_price: float = 0, contract_type: str = 'spot', expiry_date: str = '', data: dict | None = None) -> str`¶

`get_positions(commodity: str = '', limit: int = 100) -> list[dict]`¶

`update_position_price(position_id: str, current_price: float) -> bool`¶

`close_position(position_id: str, close_price: float) -> bool`¶

`record_consumption(facility_id: str, period: str, consumption_kwh: float = 0, peak_demand_kw: float = 0, energy_cost: float = 0, demand_cost: float = 0, facility_type: str = '', data: dict | None = None) -> str`¶

`get_consumption(facility_id: str = '', period: str = '', limit: int = 100) -> list[dict]`¶

`create_storage_system(name: str, technology: str = 'li_ion', capacity_kwh: float = 0, power_kw: float = 0, location: str = '', data: dict | None = None) -> str`¶

`get_storage_system(system_id: str) -> dict | None`¶

`list_storage_systems(technology: str = '', status: str = '') -> list[dict]`¶

`update_storage_health(system_id: str, soh_pct: float, cycles_completed: int) -> bool`¶

`record_emissions(period: str, scope: str = 'scope1', emissions_tco2: float = 0, source: str = '', carbon_intensity: float = 0, data: dict | None = None) -> str`¶

`get_emissions(period: str = '', scope: str = '', limit: int = 100) -> list[dict]`¶

`get_total_emissions(period: str = '', scope: str = '') -> float`¶

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

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