Job Accountant¶

job_accountant — G6 Accountant job agent.

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

Overview¶

Domain-specialist job agent that automates the full accounting cycle for an organisation. Handles double-entry bookkeeping, chart-of-accounts management, trial-balance reconciliation, GAAP/IFRS/SOX compliance checking, financial-statement generation (balance sheet, income statement, cash-flow), ASC 606 revenue-recognition validation, depreciation scheduling, tax-provision computation, deferred-tax analysis, and AR/AP ageing — all within G6's safety-bounded, audit-trailed JobAgentBlock framework.

Pilot-ready accounting assistant, not production sign-off

job_accountant is suitable for launch pilots, local MCP workflows, and assisted accounting review where a qualified person remains accountable for the final decision. Focused MCP, reconciliation, and accounting-correctness tests pass, and the component exposes the implemented accountant operations through MCP. However, the full tests/mvp/job_accountant suite currently collects 510 tests and exceeded the local 5-minute verification window during review, so do not treat this component as fully production-cleared for unattended posting, filings, audit opinions, tax positions, or regulated accounting sign-off. Keep human review enabled for period close, filings, audit work, and any material ledger changes.

When to use:

Preparing period-end financial statements and closing journal entries
Checking a ledger for GAAP or IFRS compliance flags before external audit
Computing depreciation schedules or tax provisions for a fixed-asset register
Automating variance analysis and ratio evaluation across fiscal periods

Example:

from mvp.job_accountant import JobAccountantBlock, JobAccountantInput

block = JobAccountantBlock()
result = block.infer(JobAccountantInput(
    task="Prepare Q1 income statement and flag any SOX compliance issues",
    fiscal_period="2026-Q1",
    standards=["GAAP", "SOX"],
    materiality_threshold="5000",
))
# result.ok → True; result.value → JobAccountantOutput with statements, compliance_flags

Works well with: job_framework, job_tax, job_finance

Public API¶

`JobAccountantBlock(JobAgentBlock)`¶

G6 Accountant job agent — MCP-delegated block.

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

`AccountingDecision`¶

Validated op-classification + risk-posture verdict.

Field	Type	Default
`op`	`str`	required
`baseline_op`	`str`	required
`posture`	`str`	required
`baseline_posture`	`str`	required
`requires_review`	`bool`	False
`degraded`	`bool`	False
`llm_used`	`bool`	False
`reasons`	`tuple[str, ...]`	()
`rationale`	`str`	''
`detail`	`str`	''
`signals`	`tuple[str, ...]`	()
`raw_response`	`str`	''
`baseline_regulated`	`bool`	False

Methods:

`human_review() -> bool`¶

`escalated() -> bool`¶

`op_refined() -> bool`¶

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

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

`JobAccountantPlanner`¶

Runtime-first facade with deterministic fallback + one-way posture clamp.

Field	Type	Default
`runtime`	`AccountantRuntime \\| None`	None
`last_decision`	`AccountingDecision \\| None`	field(default=None, init=False)
`last_llm_used`	`bool`	field(default=False, init=False)
`last_fallback_reason`	`str`	field(default='', init=False)
`last_grounding`	`dict[str, Any]`	field(default_factory=lambda: {'injected': False, 'fact_count': 0, 'source_count': 0, 'coverage': 'none'}, init=False)

Methods:

`classify(task: str, materiality_threshold: Any = None, materiality_amount: Any = None, context: dict[str, Any] | None = None) -> AccountingDecision`¶

`JobAccountantInput(JobInput)`¶

Input for the accountant job agent.

`JobAccountantOutput(JobOutput)`¶

Output from the accountant job agent.

`JobAccountantMCPBlock(AIBlock[MCPJobAccountantInput, MCPJobAccountantOutput, dict])`¶

29-op MCP block for the Accountant job agent.

Field	Type	Default
`name`	`str`	'job_accountant_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: MCPJobAccountantInput) -> Result[MCPJobAccountantOutput]`¶

`MCPJobAccountantInput(BaseModel)`¶

Input to JobAccountantMCPBlock — 29-op dispatch.

Field	Type	Default
`op`	Literal['prepare_statement', 'reconcile_accounts', 'process_transaction', 'consolidate_entities', 'manage_ledger', 'file_report', 'close_period', 'analyze_variance', 'evaluate_ratios', 'assess_compliance', 'benchmark_performance', 'audit_entries', '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', 'assess_task', '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`	''
`request_id`	`str`	''
`task_id`	`str`	''
`run_id`	`str`	''

`MCPJobAccountantOutput(BaseModel)`¶

Output from JobAccountantMCPBlock.

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`	`CompletionState`	'qualified-draft'
`warning_card`	`dict[str, Any]`	Field(default_factory=dict)
`evidence`	`dict[str, Any]`	Field(default_factory=dict)
`request_id`	`str`	''
`task_id`	`str`	''
`run_id`	`str`	''
`classified_op`	`str \\| None`	None
`risk_posture`	`str \\| None`	None
`human_review_required`	`bool`	False
`llm_used`	`bool`	False
`agentic_evidence`	`dict[str, Any] \\| None`	None

`AccountantStore(JobStore)`¶

SQLite store for the Accountant job agent.

Constructor:

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

Functions¶

`deterministic_accounting_decision(task: str, materiality_threshold: Any = None, materiality_amount: Any = None) -> AccountingDecision`¶

Demoted-real STRUCTURAL baseline (the zero-LLM floor).

`build_default_planner() -> 'JobAccountantPlanner | None'`¶

Build the on-by-default agentic planner, or None when suppressed.

`grounded_classify_accounting(task: str, materiality_threshold: Any = None, materiality_amount: Any = None, planner: JobAccountantPlanner | None = None) -> AccountingDecision`¶

The ONE shared grounded chokepoint for EVERY job_accountant classify decision.

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

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

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

Run grounded four-valued QA over an accountant output.

MCP Tools¶

Operation	Source
`verified`	accountant_mcp
`qualified-draft`	accountant_mcp
`blocked-escalated`	accountant_mcp
`prepare_statement`	accountant_mcp
`reconcile_accounts`	accountant_mcp
`process_transaction`	accountant_mcp
`consolidate_entities`	accountant_mcp
`manage_ledger`	accountant_mcp
`file_report`	accountant_mcp
`close_period`	accountant_mcp
`analyze_variance`	accountant_mcp
`evaluate_ratios`	accountant_mcp
`assess_compliance`	accountant_mcp
`benchmark_performance`	accountant_mcp
`audit_entries`	accountant_mcp
`create_proposal`	accountant_mcp
`review_deliverable`	accountant_mcp
`delegate_task`	accountant_mcp
`report_status`	accountant_mcp
`request_feedback`	accountant_mcp
`store_artifact`	accountant_mcp
`retrieve_artifact`	accountant_mcp
`list_artifacts`	accountant_mcp
`search_artifacts`	accountant_mcp
`archive`	accountant_mcp
`plan_sprint`	accountant_mcp
`track_progress`	accountant_mcp
`reflect_on_outcome`	accountant_mcp
`get_capabilities`	accountant_mcp
`info`	accountant_mcp
`assess_task`	accountant_mcp
`list_patterns`	accountant_mcp

Job Accountant¶

Overview¶

Public API¶

JobAccountantBlock(JobAgentBlock)¶

AccountingDecision¶

human_review() -> bool¶

escalated() -> bool¶

op_refined() -> bool¶

agentic_evidence() -> dict[str, Any]¶

to_dict() -> dict[str, Any]¶

JobAccountantPlanner¶

classify(task: str, materiality_threshold: Any = None, materiality_amount: Any = None, context: dict[str, Any] | None = None) -> AccountingDecision¶

JobAccountantInput(JobInput)¶

JobAccountantOutput(JobOutput)¶

JobAccountantMCPBlock(AIBlock[MCPJobAccountantInput, MCPJobAccountantOutput, dict])¶

infer(data: MCPJobAccountantInput) -> Result[MCPJobAccountantOutput]¶

MCPJobAccountantInput(BaseModel)¶

MCPJobAccountantOutput(BaseModel)¶

AccountantStore(JobStore)¶

Functions¶

deterministic_accounting_decision(task: str, materiality_threshold: Any = None, materiality_amount: Any = None) -> AccountingDecision¶

build_default_planner() -> 'JobAccountantPlanner | None'¶

grounded_classify_accounting(task: str, materiality_threshold: Any = None, materiality_amount: Any = None, planner: JobAccountantPlanner | None = None) -> AccountingDecision¶

assemble_review_text(output: Any) -> str¶

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

MCP Tools¶

`JobAccountantBlock(JobAgentBlock)`¶

`AccountingDecision`¶

`human_review() -> bool`¶

`escalated() -> bool`¶

`op_refined() -> bool`¶

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

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

`JobAccountantPlanner`¶

`classify(task: str, materiality_threshold: Any = None, materiality_amount: Any = None, context: dict[str, Any] | None = None) -> AccountingDecision`¶

`JobAccountantInput(JobInput)`¶

`JobAccountantOutput(JobOutput)`¶

`JobAccountantMCPBlock(AIBlock[MCPJobAccountantInput, MCPJobAccountantOutput, dict])`¶

`infer(data: MCPJobAccountantInput) -> Result[MCPJobAccountantOutput]`¶

`MCPJobAccountantInput(BaseModel)`¶

`MCPJobAccountantOutput(BaseModel)`¶

`AccountantStore(JobStore)`¶

`deterministic_accounting_decision(task: str, materiality_threshold: Any = None, materiality_amount: Any = None) -> AccountingDecision`¶

`build_default_planner() -> 'JobAccountantPlanner | None'`¶

`grounded_classify_accounting(task: str, materiality_threshold: Any = None, materiality_amount: Any = None, planner: JobAccountantPlanner | None = None) -> AccountingDecision`¶

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

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