Job Agriculture¶

job_agriculture — G6 Agriculture job agent.

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

Overview¶

Domain-specialist job agent for agricultural operations. Plans crop cycles, monitors field and climate conditions, optimises yield predictions, manages livestock schedules, forecasts harvests, and runs soil analysis within G6's safety-bounded, audit-trailed JobAgentBlock framework.

When to use:

Planning seasonal crop rotations and coordinating planting schedules across multiple fields
Monitoring real-time sensor data for soil moisture, temperature, and nutrient levels
Forecasting harvest volumes and logistics requirements ahead of the growing season
Optimising irrigation or fertilisation decisions from historical yield records

Example:

from mvp.job_agriculture import JobAgricultureBlock, JobAgricultureInput

block = JobAgricultureBlock()
result = block.infer(JobAgricultureInput(
    task="Plan the spring wheat crop cycle for field blocks A1-A4 given last season's soil reports",
    context={"region": "Murray-Darling Basin"},
    parameters={
        "crop": "wheat",
        "fields": [
            {"field_id": "A1", "current_crop": "canola", "acres": 120},
            {"field_id": "A2", "current_crop": "fallow", "acres": 95},
        ],
        "target_yield_t_ha": 4.5,
    },
))
# result.ok -> True; result.value -> JobAgricultureOutput with result,
# completion_state, warning_card, and agentic_evidence.

Common parameters:

plan_crop_cycle: crop, season, year, fields, rotation_history, soil_temp_f, last_frost_date, current_date, daily_temps
analyze_soil: field_id, crop, ph, n_ppm, p_ppm, k_ppm, organic_matter_pct, sand_pct, silt_pct, clay_pct
forecast_harvest: crop, field_id, acres, current_moisture_pct, daily_temps, historical_yields, market_price

Reliability envelope: deterministic local success reports completion_state: verified; degraded but useful fallback reports qualified-draft with warning_card; refusal, not-found, or failure reports blocked-escalated. MCP outputs preserve the legacy JSON string result and add structured_result for direct structured access.

Works well with: job_framework, job_scientist, job_logistics

Public API¶

`JobAgricultureBlock(JobAgentBlock)`¶

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

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

`JobAgricultureInput(JobInput)`¶

Input for the Agriculture job agent.

`JobAgricultureOutput(JobOutput)`¶

`JobAgricultureMCPBlock(AIBlock[MCPJobAgricultureInput, MCPJobAgricultureOutput, dict])`¶

26-op MCP block for the Agriculture job agent.

Field	Type	Default
`name`	`str`	'job_agriculture_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: MCPJobAgricultureInput) -> Result[MCPJobAgricultureOutput]`¶

`MCPJobAgricultureInput(BaseModel)`¶

Input to JobAgricultureMCPBlock -- 26-op dispatch.

Field	Type	Default
`op`	`Literal['plan_crop_cycle', 'monitor_conditions', 'optimize_yield', 'manage_livestock', 'forecast_harvest', 'analyze_soil', 'evaluate_weather', 'assess_pest_risk', 'benchmark_productivity', 'audit_sustainability', '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`	''

`MCPJobAgricultureOutput(BaseModel)`¶

Output from JobAgricultureMCPBlock.

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

`AgricultureStore(JobStore)`¶

SQLite store for the Agriculture job agent.

Constructor:

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

Methods:

`save_field(field_id: str = '', name: str = '', acres: float = 0, soil_type: str = '', current_crop: str = '', location: str = '', status: str = 'active', data: dict | None = None) -> str`¶

`get_field(field_id: str) -> dict | None`¶

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

`save_crop_plan(field_id: str = '', crop: str = '', season: str = '', year: int = 0, target_yield: float = 0, seed_rate: float = 0, planting_date: str = '', status: str = 'planned', data: dict | None = None) -> str`¶

`get_crop_plans(field_id: str = '', year: int = 0) -> list[dict]`¶

`save_soil_test(field_id: str = '', test_date: str = '', ph: float = 0, organic_matter_pct: float = 0, n_ppm: float = 0, p_ppm: float = 0, k_ppm: float = 0, cec_meq: float = 0, data: dict | None = None) -> str`¶

`get_soil_tests(field_id: str) -> list[dict]`¶

`save_irrigation(field_id: str = '', application_date: str = '', system_type: str = '', amount_mm: float = 0, et_mm: float = 0, efficiency: float = 0, data: dict | None = None) -> str`¶

`get_irrigation_records(field_id: str) -> list[dict]`¶

`save_livestock(species: str = '', herd_id: str = '', head_count: int = 0, category: str = '', data: dict | None = None) -> str`¶

`get_livestock(species: str = '') -> list[dict]`¶

`save_harvest(field_id: str = '', crop: str = '', harvest_date: str = '', yield_per_acre: float = 0, total_yield: float = 0, moisture_pct: float = 0, quality_grade: str = '', data: dict | None = None) -> str`¶

`get_harvests(field_id: str = '', crop: str = '') -> list[dict]`¶

`save_weather(station: str = '', record_date: str = '', t_max_f: float = 0, t_min_f: float = 0, rainfall_mm: float = 0, humidity_pct: float = 0, wind_mph: float = 0, data: dict | None = None) -> str`¶

`get_weather(station: str = '', limit: int = 30) -> list[dict]`¶

`save_financial(field_id: str = '', category: str = '', description: str = '', amount: float = 0, record_type: str = 'expense', record_date: str = '', data: dict | None = None) -> str`¶

`get_financials(field_id: str = '', record_type: str = '') -> list[dict]`¶

`get_financial_summary(field_id: str = '') -> dict`¶

Get revenue/expense summary for a field or entire farm.

Functions¶

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

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

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

Run grounded four-valued QA over an agriculture output.

MCP Tools¶

Operation	Source
`plan_crop_cycle`	agriculture_mcp
`monitor_conditions`	agriculture_mcp
`optimize_yield`	agriculture_mcp
`manage_livestock`	agriculture_mcp
`forecast_harvest`	agriculture_mcp
`analyze_soil`	agriculture_mcp
`evaluate_weather`	agriculture_mcp
`assess_pest_risk`	agriculture_mcp
`benchmark_productivity`	agriculture_mcp
`audit_sustainability`	agriculture_mcp
`create_proposal`	agriculture_mcp
`review_deliverable`	agriculture_mcp
`delegate_task`	agriculture_mcp
`report_status`	agriculture_mcp
`request_feedback`	agriculture_mcp
`store_artifact`	agriculture_mcp
`retrieve_artifact`	agriculture_mcp
`list_artifacts`	agriculture_mcp
`search_artifacts`	agriculture_mcp
`archive`	agriculture_mcp
`plan_sprint`	agriculture_mcp
`track_progress`	agriculture_mcp
`reflect_on_outcome`	agriculture_mcp
`list_patterns`	agriculture_mcp
`get_capabilities`	agriculture_mcp
`info`	agriculture_mcp

Job Agriculture¶

Overview¶

Public API¶

JobAgricultureBlock(JobAgentBlock)¶

JobAgricultureInput(JobInput)¶

JobAgricultureOutput(JobOutput)¶

JobAgricultureMCPBlock(AIBlock[MCPJobAgricultureInput, MCPJobAgricultureOutput, dict])¶

infer(data: MCPJobAgricultureInput) -> Result[MCPJobAgricultureOutput]¶

MCPJobAgricultureInput(BaseModel)¶

MCPJobAgricultureOutput(BaseModel)¶

AgricultureStore(JobStore)¶

save_field(field_id: str = '', name: str = '', acres: float = 0, soil_type: str = '', current_crop: str = '', location: str = '', status: str = 'active', data: dict | None = None) -> str¶

get_field(field_id: str) -> dict | None¶

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

save_crop_plan(field_id: str = '', crop: str = '', season: str = '', year: int = 0, target_yield: float = 0, seed_rate: float = 0, planting_date: str = '', status: str = 'planned', data: dict | None = None) -> str¶

get_crop_plans(field_id: str = '', year: int = 0) -> list[dict]¶

save_soil_test(field_id: str = '', test_date: str = '', ph: float = 0, organic_matter_pct: float = 0, n_ppm: float = 0, p_ppm: float = 0, k_ppm: float = 0, cec_meq: float = 0, data: dict | None = None) -> str¶

get_soil_tests(field_id: str) -> list[dict]¶

save_irrigation(field_id: str = '', application_date: str = '', system_type: str = '', amount_mm: float = 0, et_mm: float = 0, efficiency: float = 0, data: dict | None = None) -> str¶

get_irrigation_records(field_id: str) -> list[dict]¶

save_livestock(species: str = '', herd_id: str = '', head_count: int = 0, category: str = '', data: dict | None = None) -> str¶

get_livestock(species: str = '') -> list[dict]¶

save_harvest(field_id: str = '', crop: str = '', harvest_date: str = '', yield_per_acre: float = 0, total_yield: float = 0, moisture_pct: float = 0, quality_grade: str = '', data: dict | None = None) -> str¶

get_harvests(field_id: str = '', crop: str = '') -> list[dict]¶

save_weather(station: str = '', record_date: str = '', t_max_f: float = 0, t_min_f: float = 0, rainfall_mm: float = 0, humidity_pct: float = 0, wind_mph: float = 0, data: dict | None = None) -> str¶

get_weather(station: str = '', limit: int = 30) -> list[dict]¶

save_financial(field_id: str = '', category: str = '', description: str = '', amount: float = 0, record_type: str = 'expense', record_date: str = '', data: dict | None = None) -> str¶

get_financials(field_id: str = '', record_type: str = '') -> list[dict]¶

get_financial_summary(field_id: str = '') -> dict¶