Job Logistics¶

job_logistics — G6 Logistics job agent.

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

Overview¶

Domain-specialist job agent for supply-chain and logistics professionals. Optimises delivery routes, tracks shipment status, manages inventory levels, schedules last-mile deliveries, audits warehouse operations, and analyses throughput and on-time delivery metrics within G6's safety-bounded, audit-trailed JobAgentBlock framework.

This component exposes 26 MCP tools and returns canonical reliability envelope fields: completion_state (verified, qualified-draft, or blocked-escalated), warning_card, and evidence. Compliance-floor markers for hazmat/dangerous goods, customs/export controls, and cold-chain scenarios are surfaced through warning_card.code == "G6_E_COMPLIANCE_FLOOR" and require human review.

Production use requires a qualifying reviewer_signature because the block contract declares domain_expert_review. The component does not provide live carrier/TMS API tracking or live freight-exchange pricing; shipment state is local SQLite data and cost/ETA outputs are advisory calculations unless the caller supplies current source data.

When to use:

Optimising multi-stop delivery routes under vehicle capacity and time-window constraints
Tracking shipment status from supplied or locally stored data and generating exception alerts
Auditing warehouse inventory for discrepancies between system records and physical counts
Analysing network throughput to identify bottlenecks and recommend capacity changes

Example:

from mvp.job_logistics import JobLogisticsBlock, JobLogisticsInput

block = JobLogisticsBlock()
result = block.infer(JobLogisticsInput(
    task="Optimise delivery routes for 42 drops across Greater Sydney for Tuesday 8 April, minimising total distance",
    context={"fleet_size": 6, "vehicle_capacity_kg": 1200, "depot": "Wetherill Park"},
))
# result.ok -> True; inspect result.value.completion_state, warning_card, and evidence

Works well with: job_framework, job_manufacturing, job_manager

Public API¶

`LogisticsDecision`¶

Validated op-classification decision for a logistics task.

Field	Type	Default
`op`	`str`	required
`recommendation`	`str`	''
`rationale`	`str`	''
`signals`	`list[str]`	field(default_factory=list)
`citations`	`list[str]`	field(default_factory=list)
`compliance_flags`	`list[str]`	field(default_factory=list)
`confidence`	`float`	0.0
`completion_state`	`str`	'qualified-draft'
`degraded`	`bool`	False
`raw_response`	`str`	''

`LogisticsPlanner`¶

Runtime-first facade with deterministic fallback + one-way compliance floor.

Constructor:

Parameter	Type	Default
`runtime`	`LogisticsRuntime \\| None`	None

Methods:

`classify(task: str, context: dict | None = None) -> LogisticsDecision`¶

`JobLogisticsBlock(JobAgentBlock)`¶

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

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

Methods:

`infer(data: Any) -> Result[JobLogisticsOutput]`¶

`LogisticsPatternRuntime`¶

Stateless executable mechanisms for logistics' applied patterns.

Methods:

`verify_against_floor(agentic_op: str, floor_op: str, compliance_flags: list[str] | tuple[str, ...], llm_used: bool = True, confidence: float | None = None) -> LogisticsPatternReview`¶

output-verification-loop + human-in-loop-approval-framework.

`requires_review(decision: Any) -> bool`¶

human-in-loop-approval-framework: does this decision need review?

`JobLogisticsInput(JobInput)`¶

Input for the Logistics job agent.

`JobLogisticsOutput(JobOutput)`¶

`LogisticsSkillCatalog`¶

Maps each applied pattern slug to a logistics op-classification skill.

Methods:

`list_skills() -> list[LogisticsSkill]`¶

`executable_skills() -> list[LogisticsSkill]`¶

`get(slug: str) -> LogisticsSkill | None`¶

`JobLogisticsMCPBlock(AIBlock[MCPJobLogisticsInput, MCPJobLogisticsOutput, dict])`¶

26-op MCP block for the Logistics job agent.

Field	Type	Default
`name`	`str`	'job_logistics_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: MCPJobLogisticsInput) -> Result[MCPJobLogisticsOutput]`¶

`MCPJobLogisticsInput(BaseModel)`¶

Input to JobLogisticsMCPBlock — 26-op dispatch.

Field	Type	Default
`op`	`Literal['optimize_route', 'track_shipment', 'manage_inventory', 'schedule_delivery', 'audit_warehouse', 'analyze_throughput', 'evaluate_carrier', 'assess_capacity', 'benchmark_cost', 'audit_compliance', '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`	''

`MCPJobLogisticsOutput(BaseModel)`¶

Output from JobLogisticsMCPBlock.

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

`LogisticsStore(JobStore)`¶

SQLite store for the Logistics job agent.

Constructor:

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

Methods:

`create_shipment(origin: str, destination: str, carrier_id: str = '', mode: str = 'truck', weight_lbs: float = 0, tracking_number: str = '', estimated_arrival: str = '', cost: float = 0, data: dict | None = None) -> str`¶

`get_shipment(shipment_id: str) -> dict | None`¶

`update_shipment_status(shipment_id: str, status: str, actual_arrival: str = '') -> bool`¶

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

`add_inventory_item(sku: str, warehouse_id: str = 'WH01', quantity: float = 0, aisle: int = 0, bay: int = 0, level: int = 0, unit_value: float = 0, data: dict | None = None) -> str`¶

`get_inventory(warehouse_id: str = '', sku: str = '') -> list[dict]`¶

`update_inventory_quantity(item_id: str, quantity: float) -> bool`¶

`create_route(origin: str, destination: str, distance_miles: float = 0, mode: str = 'truck', transit_hours: float = 0, stops: list | None = None, cost_estimate: float = 0, data: dict | None = None) -> str`¶

`get_route(route_id: str) -> dict | None`¶

`list_routes(origin: str = '', destination: str = '') -> list[dict]`¶

`add_carrier(name: str, modes: str = 'truck', on_time_count: int = 0, total_deliveries: int = 0, damage_count: int = 0, claim_count: int = 0, rating: str = '', data: dict | None = None) -> str`¶

`get_carrier(carrier_id: str) -> dict | None`¶

`list_carriers() -> list[dict]`¶

`create_customs_record(shipment_id: str, country_of_origin: str = '', hs_code: str = '', declared_value: float = 0, duty_amount: float = 0, fta_applied: bool = False, compliance_status: str = 'pending', data: dict | None = None) -> str`¶

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

`create_delivery_schedule(origin: str, destination: str, stops_json: str = '[]', total_distance: float = 0, estimated_cost: float = 0, delivery_count: int = 0, mode: str = 'truck', data: dict | None = None) -> str`¶

`list_delivery_schedules(limit: int = 50) -> list[dict]`¶

`create_warehouse_audit(warehouse_id: str, utilization_pct: float = 0, throughput_uph: float = 0, slotting_efficiency: float = 0, data: dict | None = None) -> str`¶

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

`update_carrier_rating(carrier_id: str, rating: str) -> bool`¶

`update_carrier_stats(carrier_id: str, on_time_count: int | None = None, total_deliveries: int | None = None, damage_count: int | None = None, claim_count: int | None = None) -> bool`¶

`search_carriers(name_pattern: str = '', min_rating: str = '') -> list[dict]`¶

`update_customs_compliance(record_id: str, compliance_status: str) -> bool`¶

`get_customs_summary() -> dict`¶

Return compliance status counts across all customs records.

`update_inventory_status(item_id: str, status: str) -> bool`¶

`get_inventory_value(warehouse_id: str = '') -> dict`¶

Total inventory value for a warehouse (or all).

`get_low_stock_items(threshold: float = 10) -> list[dict]`¶

Items with quantity below threshold.

`search_routes(mode: str = '', min_distance: float = 0, max_distance: float = 999999) -> list[dict]`¶

`get_shipment_summary() -> dict`¶

Shipment counts by status.

`get_recent_activity(limit: int = 20) -> list[dict]`¶

Recent history entries for dashboards.

Functions¶

`summarize_job_logistics_agentic_evidence(decisions: list[dict[str, Any]]) -> dict[str, Any]`¶

Summarise runtime-vs-fallback logistics decisions with path redaction.

`agentic_planner_enabled(default_enabled: bool) -> bool`¶

Decide whether the agentic logistics planner should be used.

`compliance_floor(task: str, context: dict | None = None) -> list[str]`¶

Deterministic, task-independent scan for compliance/safety markers.

`deterministic_classify(task: str, context: dict | None = None) -> LogisticsDecision`¶

Demoted real keyword op-classifier + the deterministic compliance floor.

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

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

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

Run grounded four-valued QA over a logistics output.

`applied_agentic_patterns() -> list[dict[str, Any]]`¶

Return compact metadata for the logistics-applied vendored patterns.

MCP Tools¶

Operation	Source
`optimize_route`	logistics_mcp
`track_shipment`	logistics_mcp
`manage_inventory`	logistics_mcp
`schedule_delivery`	logistics_mcp
`audit_warehouse`	logistics_mcp
`analyze_throughput`	logistics_mcp
`evaluate_carrier`	logistics_mcp
`assess_capacity`	logistics_mcp
`benchmark_cost`	logistics_mcp
`audit_compliance`	logistics_mcp
`create_proposal`	logistics_mcp
`review_deliverable`	logistics_mcp
`delegate_task`	logistics_mcp
`report_status`	logistics_mcp
`request_feedback`	logistics_mcp
`store_artifact`	logistics_mcp
`retrieve_artifact`	logistics_mcp
`list_artifacts`	logistics_mcp
`search_artifacts`	logistics_mcp
`archive`	logistics_mcp
`plan_sprint`	logistics_mcp
`track_progress`	logistics_mcp
`reflect_on_outcome`	logistics_mcp
`get_capabilities`	logistics_mcp
`info`	logistics_mcp
`list_patterns`	logistics_mcp

Job Logistics¶

Overview¶

Public API¶

LogisticsDecision¶

LogisticsPlanner¶

classify(task: str, context: dict | None = None) -> LogisticsDecision¶

JobLogisticsBlock(JobAgentBlock)¶

infer(data: Any) -> Result[JobLogisticsOutput]¶

LogisticsPatternRuntime¶

verify_against_floor(agentic_op: str, floor_op: str, compliance_flags: list[str] | tuple[str, ...], llm_used: bool = True, confidence: float | None = None) -> LogisticsPatternReview¶

requires_review(decision: Any) -> bool¶

JobLogisticsInput(JobInput)¶

JobLogisticsOutput(JobOutput)¶

LogisticsSkillCatalog¶

list_skills() -> list[LogisticsSkill]¶

executable_skills() -> list[LogisticsSkill]¶

get(slug: str) -> LogisticsSkill | None¶

JobLogisticsMCPBlock(AIBlock[MCPJobLogisticsInput, MCPJobLogisticsOutput, dict])¶

infer(data: MCPJobLogisticsInput) -> Result[MCPJobLogisticsOutput]¶

MCPJobLogisticsInput(BaseModel)¶

MCPJobLogisticsOutput(BaseModel)¶

LogisticsStore(JobStore)¶

create_shipment(origin: str, destination: str, carrier_id: str = '', mode: str = 'truck', weight_lbs: float = 0, tracking_number: str = '', estimated_arrival: str = '', cost: float = 0, data: dict | None = None) -> str¶

get_shipment(shipment_id: str) -> dict | None¶

update_shipment_status(shipment_id: str, status: str, actual_arrival: str = '') -> bool¶

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

add_inventory_item(sku: str, warehouse_id: str = 'WH01', quantity: float = 0, aisle: int = 0, bay: int = 0, level: int = 0, unit_value: float = 0, data: dict | None = None) -> str¶

get_inventory(warehouse_id: str = '', sku: str = '') -> list[dict]¶

update_inventory_quantity(item_id: str, quantity: float) -> bool¶

create_route(origin: str, destination: str, distance_miles: float = 0, mode: str = 'truck', transit_hours: float = 0, stops: list | None = None, cost_estimate: float = 0, data: dict | None = None) -> str¶

get_route(route_id: str) -> dict | None¶

list_routes(origin: str = '', destination: str = '') -> list[dict]¶

add_carrier(name: str, modes: str = 'truck', on_time_count: int = 0, total_deliveries: int = 0, damage_count: int = 0, claim_count: int = 0, rating: str = '', data: dict | None = None) -> str¶

get_carrier(carrier_id: str) -> dict | None¶

list_carriers() -> list[dict]¶

create_customs_record(shipment_id: str, country_of_origin: str = '', hs_code: str = '', declared_value: float = 0, duty_amount: float = 0, fta_applied: bool = False, compliance_status: str = 'pending', data: dict | None = None) -> str¶

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

create_delivery_schedule(origin: str, destination: str, stops_json: str = '[]', total_distance: float = 0, estimated_cost: float = 0, delivery_count: int = 0, mode: str = 'truck', data: dict | None = None) -> str¶

list_delivery_schedules(limit: int = 50) -> list[dict]¶

create_warehouse_audit(warehouse_id: str, utilization_pct: float = 0, throughput_uph: float = 0, slotting_efficiency: float = 0, data: dict | None = None) -> str¶

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

update_carrier_rating(carrier_id: str, rating: str) -> bool¶

update_carrier_stats(carrier_id: str, on_time_count: int | None = None, total_deliveries: int | None = None, damage_count: int | None = None, claim_count: int | None = None) -> bool¶

search_carriers(name_pattern: str = '', min_rating: str = '') -> list[dict]¶

update_customs_compliance(record_id: str, compliance_status: str) -> bool¶

get_customs_summary() -> dict¶

update_inventory_status(item_id: str, status: str) -> bool¶

get_inventory_value(warehouse_id: str = '') -> dict¶

get_low_stock_items(threshold: float = 10) -> list[dict]¶

search_routes(mode: str = '', min_distance: float = 0, max_distance: float = 999999) -> list[dict]¶

get_shipment_summary() -> dict¶

get_recent_activity(limit: int = 20) -> list[dict]¶

Functions¶

summarize_job_logistics_agentic_evidence(decisions: list[dict[str, Any]]) -> dict[str, Any]¶

agentic_planner_enabled(default_enabled: bool) -> bool¶

compliance_floor(task: str, context: dict | None = None) -> list[str]¶

deterministic_classify(task: str, context: dict | None = None) -> LogisticsDecision¶

assemble_review_text(output: Any) -> str¶

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

applied_agentic_patterns() -> list[dict[str, Any]]¶

MCP Tools¶

`LogisticsDecision`¶

`LogisticsPlanner`¶

`classify(task: str, context: dict | None = None) -> LogisticsDecision`¶

`JobLogisticsBlock(JobAgentBlock)`¶

`infer(data: Any) -> Result[JobLogisticsOutput]`¶

`LogisticsPatternRuntime`¶

`verify_against_floor(agentic_op: str, floor_op: str, compliance_flags: list[str] | tuple[str, ...], llm_used: bool = True, confidence: float | None = None) -> LogisticsPatternReview`¶

`requires_review(decision: Any) -> bool`¶

`JobLogisticsInput(JobInput)`¶

`JobLogisticsOutput(JobOutput)`¶

`LogisticsSkillCatalog`¶

`list_skills() -> list[LogisticsSkill]`¶

`executable_skills() -> list[LogisticsSkill]`¶

`get(slug: str) -> LogisticsSkill | None`¶

`JobLogisticsMCPBlock(AIBlock[MCPJobLogisticsInput, MCPJobLogisticsOutput, dict])`¶

`infer(data: MCPJobLogisticsInput) -> Result[MCPJobLogisticsOutput]`¶

`MCPJobLogisticsInput(BaseModel)`¶

`MCPJobLogisticsOutput(BaseModel)`¶

`LogisticsStore(JobStore)`¶

`create_shipment(origin: str, destination: str, carrier_id: str = '', mode: str = 'truck', weight_lbs: float = 0, tracking_number: str = '', estimated_arrival: str = '', cost: float = 0, data: dict | None = None) -> str`¶

`get_shipment(shipment_id: str) -> dict | None`¶

`update_shipment_status(shipment_id: str, status: str, actual_arrival: str = '') -> bool`¶

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

`add_inventory_item(sku: str, warehouse_id: str = 'WH01', quantity: float = 0, aisle: int = 0, bay: int = 0, level: int = 0, unit_value: float = 0, data: dict | None = None) -> str`¶

`get_inventory(warehouse_id: str = '', sku: str = '') -> list[dict]`¶

`update_inventory_quantity(item_id: str, quantity: float) -> bool`¶

`create_route(origin: str, destination: str, distance_miles: float = 0, mode: str = 'truck', transit_hours: float = 0, stops: list | None = None, cost_estimate: float = 0, data: dict | None = None) -> str`¶

`get_route(route_id: str) -> dict | None`¶

`list_routes(origin: str = '', destination: str = '') -> list[dict]`¶

`add_carrier(name: str, modes: str = 'truck', on_time_count: int = 0, total_deliveries: int = 0, damage_count: int = 0, claim_count: int = 0, rating: str = '', data: dict | None = None) -> str`¶

`get_carrier(carrier_id: str) -> dict | None`¶

`list_carriers() -> list[dict]`¶

`create_customs_record(shipment_id: str, country_of_origin: str = '', hs_code: str = '', declared_value: float = 0, duty_amount: float = 0, fta_applied: bool = False, compliance_status: str = 'pending', data: dict | None = None) -> str`¶

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

`create_delivery_schedule(origin: str, destination: str, stops_json: str = '[]', total_distance: float = 0, estimated_cost: float = 0, delivery_count: int = 0, mode: str = 'truck', data: dict | None = None) -> str`¶

`list_delivery_schedules(limit: int = 50) -> list[dict]`¶

`create_warehouse_audit(warehouse_id: str, utilization_pct: float = 0, throughput_uph: float = 0, slotting_efficiency: float = 0, data: dict | None = None) -> str`¶

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

`update_carrier_rating(carrier_id: str, rating: str) -> bool`¶

`update_carrier_stats(carrier_id: str, on_time_count: int | None = None, total_deliveries: int | None = None, damage_count: int | None = None, claim_count: int | None = None) -> bool`¶

`search_carriers(name_pattern: str = '', min_rating: str = '') -> list[dict]`¶

`update_customs_compliance(record_id: str, compliance_status: str) -> bool`¶

`get_customs_summary() -> dict`¶

`update_inventory_status(item_id: str, status: str) -> bool`¶

`get_inventory_value(warehouse_id: str = '') -> dict`¶

`get_low_stock_items(threshold: float = 10) -> list[dict]`¶

`search_routes(mode: str = '', min_distance: float = 0, max_distance: float = 999999) -> list[dict]`¶

`get_shipment_summary() -> dict`¶

`get_recent_activity(limit: int = 20) -> list[dict]`¶

`summarize_job_logistics_agentic_evidence(decisions: list[dict[str, Any]]) -> dict[str, Any]`¶

`agentic_planner_enabled(default_enabled: bool) -> bool`¶

`compliance_floor(task: str, context: dict | None = None) -> list[str]`¶

`deterministic_classify(task: str, context: dict | None = None) -> LogisticsDecision`¶

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

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

`applied_agentic_patterns() -> list[dict[str, Any]]`¶