Job Manager¶

job_manager — G6 Manager job agent.

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

Overview¶

Domain-specialist job agent for team and project managers. Sets objectives and OKRs, allocates resources across competing priorities, reviews individual and team performance, mediates conflicts, produces stakeholder reports, and analyses team capacity and velocity — providing a full management workflow within G6's safety-bounded, audit-trailed JobAgentBlock framework.

When to use:

Setting and tracking team OKRs with weekly progress check-ins and drift alerts
Allocating headcount and budget across competing project demands
Preparing structured status reports for executive or client stakeholders
Mediating team conflicts by generating structured resolution options

Example:

from mvp.job_manager import JobManagerBlock, JobManagerInput

block = JobManagerBlock()
result = block.infer(JobManagerInput(
    task="Allocate the engineering team's 80 available person-days across 5 active projects for Q2, prioritising by revenue impact",
    context={"team_size": 10, "projects": ["Phoenix", "Atlas", "Orion", "Nova", "Titan"]},
))
# result.ok → True; result.value → JobManagerOutput with result, artifacts

Works well with: job_framework, job_analyst, job_business

Source health: HRIS is currently disclosed as hris_backend="in_memory_only" and hris_live=false; team and performance outputs use caller-supplied parameters unless live HRIS evidence is explicitly present.

Reliability envelope: completion_state uses only verified, qualified-draft, or blocked-escalated. Degraded integration paths return qualified-draft with warning and evidence.

Public API¶

`JobManagerBlock(JobAgentBlock)`¶

G6 Manager job agent — Tier 1 block.

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

Methods:

`infer(data: JobManagerInput) -> Result[JobManagerOutput]`¶

`JobManagerInput(JobInput)`¶

Input for the Manager job agent.

`JobManagerOutput(JobOutput)`¶

`JobManagerMCPBlock(AIBlock[MCPJobManagerInput, MCPJobManagerOutput, dict])`¶

26-op MCP block for the Manager job agent.

Field	Type	Default
`name`	`str`	'job_manager_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: MCPJobManagerInput) -> Result[MCPJobManagerOutput]`¶

`MCPJobManagerInput(BaseModel)`¶

Input to JobManagerMCPBlock — 25-op dispatch.

Field	Type	Default
`op`	`Literal['set_objectives', 'allocate_resources', 'review_performance', 'resolve_conflict', 'report_to_stakeholders', 'analyze_team', 'evaluate_strategy', 'assess_risk', 'benchmark_kpis', 'audit_budget', '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`	''

`MCPJobManagerOutput(BaseModel)`¶

Output from JobManagerMCPBlock.

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)

`ManagerStore(JobStore)`¶

SQLite store for the Manager job agent.

Constructor:

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

Methods:

`add_team_member(name: str, role: str = '', department: str = '', skills: list | None = None, **kw: Any) -> str`¶

`get_team_member(member_id: str) -> dict | None`¶

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

`update_team_member(member_id: str, **fields: Any) -> bool`¶

`add_project(name: str, objective: str = '', budget: float = 0.0, start_date: str = '', end_date: str = '', milestones: list | None = None, data: dict | None = None) -> str`¶

`get_project(project_id: str) -> dict | None`¶

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

`update_project(project_id: str, **fields: Any) -> bool`¶

`add_performance_review(member_id: str, period: str, ratings: dict | None = None, goals: list | None = None, feedback: str = '', overall_rating: float = 0.0) -> str`¶

`get_reviews_for_member(member_id: str) -> list[dict]`¶

`add_allocation(member_id: str, project_id: str, role: str = '', allocation_pct: float = 100.0, start_date: str = '', end_date: str = '') -> str`¶

`get_allocations_for_project(project_id: str) -> list[dict]`¶

`get_allocations_for_member(member_id: str) -> list[dict]`¶

`add_meeting(title: str, meeting_type: str = 'general', attendees: list | None = None, agenda: list | None = None, decisions: list | None = None, action_items: list | None = None, meeting_date: str = '') -> str`¶

`get_meeting(meeting_id: str) -> dict | None`¶

`list_meetings(meeting_type: str = '') -> list[dict]`¶

Functions¶

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

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

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

Run grounded four-valued QA over a manager output.

MCP Tools¶

Operation	Source
`set_objectives`	manager_mcp
`allocate_resources`	manager_mcp
`review_performance`	manager_mcp
`resolve_conflict`	manager_mcp
`report_to_stakeholders`	manager_mcp
`analyze_team`	manager_mcp
`evaluate_strategy`	manager_mcp
`assess_risk`	manager_mcp
`benchmark_kpis`	manager_mcp
`audit_budget`	manager_mcp
`create_proposal`	manager_mcp
`review_deliverable`	manager_mcp
`delegate_task`	manager_mcp
`report_status`	manager_mcp
`request_feedback`	manager_mcp
`store_artifact`	manager_mcp
`retrieve_artifact`	manager_mcp
`list_artifacts`	manager_mcp
`search_artifacts`	manager_mcp
`archive`	manager_mcp
`plan_sprint`	manager_mcp
`track_progress`	manager_mcp
`reflect_on_outcome`	manager_mcp
`get_capabilities`	manager_mcp
`info`	manager_mcp
`list_patterns`	manager_mcp

Job Manager¶

Overview¶

Public API¶

JobManagerBlock(JobAgentBlock)¶

infer(data: JobManagerInput) -> Result[JobManagerOutput]¶

JobManagerInput(JobInput)¶

JobManagerOutput(JobOutput)¶

JobManagerMCPBlock(AIBlock[MCPJobManagerInput, MCPJobManagerOutput, dict])¶

infer(data: MCPJobManagerInput) -> Result[MCPJobManagerOutput]¶

MCPJobManagerInput(BaseModel)¶

MCPJobManagerOutput(BaseModel)¶

ManagerStore(JobStore)¶

add_team_member(name: str, role: str = '', department: str = '', skills: list | None = None, **kw: Any) -> str¶

get_team_member(member_id: str) -> dict | None¶

list_team_members(department: str = '', status: str = '') -> list[dict]¶

update_team_member(member_id: str, **fields: Any) -> bool¶

add_project(name: str, objective: str = '', budget: float = 0.0, start_date: str = '', end_date: str = '', milestones: list | None = None, data: dict | None = None) -> str¶

get_project(project_id: str) -> dict | None¶

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

update_project(project_id: str, **fields: Any) -> bool¶

add_performance_review(member_id: str, period: str, ratings: dict | None = None, goals: list | None = None, feedback: str = '', overall_rating: float = 0.0) -> str¶

get_reviews_for_member(member_id: str) -> list[dict]¶

add_allocation(member_id: str, project_id: str, role: str = '', allocation_pct: float = 100.0, start_date: str = '', end_date: str = '') -> str¶

get_allocations_for_project(project_id: str) -> list[dict]¶

get_allocations_for_member(member_id: str) -> list[dict]¶

add_meeting(title: str, meeting_type: str = 'general', attendees: list | None = None, agenda: list | None = None, decisions: list | None = None, action_items: list | None = None, meeting_date: str = '') -> str¶

get_meeting(meeting_id: str) -> dict | None¶

list_meetings(meeting_type: str = '') -> list[dict]¶

Functions¶

assemble_review_text(output: Any) -> str¶

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

MCP Tools¶

`JobManagerBlock(JobAgentBlock)`¶

`infer(data: JobManagerInput) -> Result[JobManagerOutput]`¶

`JobManagerInput(JobInput)`¶

`JobManagerOutput(JobOutput)`¶

`JobManagerMCPBlock(AIBlock[MCPJobManagerInput, MCPJobManagerOutput, dict])`¶

`infer(data: MCPJobManagerInput) -> Result[MCPJobManagerOutput]`¶

`MCPJobManagerInput(BaseModel)`¶

`MCPJobManagerOutput(BaseModel)`¶

`ManagerStore(JobStore)`¶

`add_team_member(name: str, role: str = '', department: str = '', skills: list | None = None, **kw: Any) -> str`¶

`get_team_member(member_id: str) -> dict | None`¶

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

`update_team_member(member_id: str, **fields: Any) -> bool`¶

`add_project(name: str, objective: str = '', budget: float = 0.0, start_date: str = '', end_date: str = '', milestones: list | None = None, data: dict | None = None) -> str`¶

`get_project(project_id: str) -> dict | None`¶

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

`update_project(project_id: str, **fields: Any) -> bool`¶

`add_performance_review(member_id: str, period: str, ratings: dict | None = None, goals: list | None = None, feedback: str = '', overall_rating: float = 0.0) -> str`¶

`get_reviews_for_member(member_id: str) -> list[dict]`¶

`add_allocation(member_id: str, project_id: str, role: str = '', allocation_pct: float = 100.0, start_date: str = '', end_date: str = '') -> str`¶

`get_allocations_for_project(project_id: str) -> list[dict]`¶

`get_allocations_for_member(member_id: str) -> list[dict]`¶

`add_meeting(title: str, meeting_type: str = 'general', attendees: list | None = None, agenda: list | None = None, decisions: list | None = None, action_items: list | None = None, meeting_date: str = '') -> str`¶

`get_meeting(meeting_id: str) -> dict | None`¶

`list_meetings(meeting_type: str = '') -> list[dict]`¶

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

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