Job It¶

job_it — G6 IT job agent.

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

Overview¶

Domain-specialist job agent for IT operations and infrastructure professionals. Deploys services, monitors system health, troubleshoots incidents, configures infrastructure-as-code, manages access controls, and analyses logs — providing a structured ITOps workflow within G6's safety-bounded, audit-trailed JobAgentBlock framework.

When to use:

Triaging production incidents by correlating logs, metrics, and recent deployment events
Deploying or scaling services in a cloud environment via structured task execution
Reviewing infrastructure configuration for security misconfigurations or drift
Analysing historical log data to surface recurring error patterns

Example:

from mvp.job_it import JobITBlock, JobITInput

block = JobITBlock()
result = block.infer(JobITInput(
    task="Diagnose the root cause of elevated 503 errors on the payments API from the last 2 hours of logs",
    context={"service": "payments-api", "environment": "prod", "log_source": "cloudwatch"},
))
# result.ok → True; result.value → JobITOutput with result, artifacts

Works well with: job_framework, job_engineer, job_manager

Production Behavior And Caveats¶

job_it is designed to be fast and deterministic on a clean MCP install. Optional LLM, debate, grounding, and web-search enrichment is off by default so ordinary tool calls do not block on unavailable model servers, network access, or paid APIs. Enable richer advisory output explicitly with G6_JOB_IT_OPTIONAL_ENRICHMENT=1 for LLM/debate paths and G6_JOB_IT_EXTERNAL_ENRICHMENT=1 for grounding/search paths; tune LLM latency with G6_JOB_IT_LLM_TIMEOUT_SECONDS.

Production integrations are best-effort. Prometheus, Grafana, Docker, and Kubernetes adapters degrade gracefully when they are not configured or reachable. Live Docker/Kubernetes deployment remains dry-run outside production unless explicitly configured for real execution.

RBAC is enforced for direct dispatch of operator/admin operations such as deployment, incident response, ticket management, access changes, and permission audits. The local MCP block seeds an internal first-run context so smoke tests and basic local workflows work immediately after install; production callers should pass real user context and roles.

benchmark_latency needs real measurements or latency_ms samples for a performance verdict. Without samples it returns an unknown tier with collection guidance rather than pretending to measure latency.

Security assessment is static and deterministic by default. It is useful for common CWE/OWASP/configuration issues, but it is not a replacement for full SAST/DAST, dependency scanning, change review, or human approval before production infrastructure changes.

Public API¶

`JobITBlock(JobAgentBlock)`¶

G6 IT job agent - MCP-delegated block.

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

`JobITInput(JobInput)`¶

Input for the IT job agent.

`JobITOutput(JobOutput)`¶

Output from the IT job agent.

`JobITMCPBlock(AIBlock[MCPJobITInput, MCPJobITOutput, dict])`¶

29-op MCP block for the IT job agent.

Field	Type	Default
`name`	`str`	'job_it_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: MCPJobITInput) -> Result[MCPJobITOutput]`¶

`MCPJobITInput(BaseModel)`¶

Input to JobITMCPBlock - 29-op dispatch.

Field	Type	Default
`op`	Literal['deploy_service', 'monitor_system', 'troubleshoot', 'configure_infra', 'manage_access', 'analyze_logs', 'evaluate_uptime', 'assess_security', 'benchmark_latency', 'audit_permissions', 'forecast_capacity', 'incident_response', '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', 'manage_tickets']	required
`task`	`str`	''
`context`	`dict[str, Any]`	Field(default_factory=dict)
`parameters`	`dict[str, Any]`	Field(default_factory=dict)
`artifact_id`	`str`	''
`query`	`str`	''

`MCPJobITOutput(BaseModel)`¶

Output from JobITMCPBlock.

Field	Type	Default
`op`	`str`	required
`completion_state`	`Literal['verified', 'qualified-draft', 'blocked-escalated']`	'qualified-draft'
`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

`ITStore(JobStore)`¶

SQLite store for the IT job agent.

Constructor:

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

Methods:

`record_metric(service_name: str, metric_name: str, value: float) -> str`¶

Store a single metric snapshot.

`get_metric_history(service_name: str, metric_name: str, limit: int = 100) -> list[dict]`¶

Retrieve recent snapshots for a (service, metric) pair.

`get_all_metrics_for_service(service_name: str, limit: int = 200) -> list[dict]`¶

Retrieve all recent metric snapshots for a given service.

`record_deployment(service_name: str, version: str, strategy: str, deploy_status: str, rollback_available: bool, previous_version: str, checks_passed: list, checks_failed: list) -> str`¶

Persist a deployment event to deployment_history.

`get_deployment_history(service_name: str, limit: int = 20) -> list[dict]`¶

Retrieve deployment history for a service, most recent first.

Functions¶

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

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

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

Run grounded four-valued QA over an IT output.

MCP Tools¶

Operation	Source
`deploy_service`	it_mcp
`monitor_system`	it_mcp
`troubleshoot`	it_mcp
`configure_infra`	it_mcp
`manage_access`	it_mcp
`analyze_logs`	it_mcp
`evaluate_uptime`	it_mcp
`assess_security`	it_mcp
`benchmark_latency`	it_mcp
`audit_permissions`	it_mcp
`forecast_capacity`	it_mcp
`incident_response`	it_mcp
`create_proposal`	it_mcp
`review_deliverable`	it_mcp
`delegate_task`	it_mcp
`report_status`	it_mcp
`request_feedback`	it_mcp
`store_artifact`	it_mcp
`retrieve_artifact`	it_mcp
`list_artifacts`	it_mcp
`search_artifacts`	it_mcp
`archive`	it_mcp
`plan_sprint`	it_mcp
`track_progress`	it_mcp
`reflect_on_outcome`	it_mcp
`list_patterns`	it_mcp
`get_capabilities`	it_mcp
`info`	it_mcp
`manage_tickets`	it_mcp

Job It¶

Overview¶

Production Behavior And Caveats¶

Public API¶

JobITBlock(JobAgentBlock)¶

JobITInput(JobInput)¶

JobITOutput(JobOutput)¶

JobITMCPBlock(AIBlock[MCPJobITInput, MCPJobITOutput, dict])¶

infer(data: MCPJobITInput) -> Result[MCPJobITOutput]¶

MCPJobITInput(BaseModel)¶

MCPJobITOutput(BaseModel)¶

ITStore(JobStore)¶

record_metric(service_name: str, metric_name: str, value: float) -> str¶

get_metric_history(service_name: str, metric_name: str, limit: int = 100) -> list[dict]¶

get_all_metrics_for_service(service_name: str, limit: int = 200) -> list[dict]¶

record_deployment(service_name: str, version: str, strategy: str, deploy_status: str, rollback_available: bool, previous_version: str, checks_passed: list, checks_failed: list) -> str¶

get_deployment_history(service_name: str, limit: int = 20) -> list[dict]¶