Job Tax¶

job_tax — G6 Tax job agent.

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

Overview¶

Domain-specialist job agent for tax professionals and advisors. Supports pilot workflows for tax-return preparation, tax-liability estimates, deduction review, refund/balance-due calculations, compliance checklists, and bracket or marginal-rate analysis within G6's safety-bounded, audit-trailed JobAgentBlock framework.

Pilot-ready, not compliance-grade tax filing

job_tax is useful for design-partner demos, reviewed workflow prototypes, and accountant-assisted analysis. It is not an unsupervised tax filing product, tax advice engine, compliance certification system, or substitute for a credentialed tax professional. Local calculations use embedded/static rule tables for selected jurisdictions and years; they can be incomplete, stale, or inapplicable to a user's facts. High-stakes operations such as filing, deduction review, and compliance assessment require human review/HITL approval. Do not rely on this component for actual filing, planning, controversy, or regulated customer claims without independent professional review and jurisdiction-specific validation.

When to use:

Drafting reviewed individual tax-return summaries with structured deduction schedules
Estimating tax liability for supported US tax years or Australian resident-income helper cases
Reviewing expense claims and deductions as a triage step before professional sign-off
Performing tax-bracket modelling for financial planning or salary-packaging discussions

Example:

from mvp.job_tax import JobTaxBlock, JobTaxInput

block = JobTaxBlock()
result = block.infer(JobTaxInput(
    task="Calculate Australian resident income tax for a reviewed planning worksheet",
    jurisdiction="AU",
    context={
        "cpa_context": True,
        "attestation": {
            "credential_id": "CPA-EXAMPLE",
            "jurisdiction": "AU",
            "issuer": "Example professional body",
        },
    },
    parameters={"gross_income": "100000", "income_year": "2025-26"},
))
# result.ok -> True; result.value -> JobTaxOutput with result, artifacts

Works well with: job_framework, job_accountant, job_finance

Public API¶

`JobTaxInput(JobInput)`¶

Input for the Tax job agent.

`JobTaxOutput(JobOutput)`¶

`JobTaxBlock(JobAgentBlock)`¶

G6 Tax job agent — Tier 1 block with MCP delegation and three safety floors.

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

`MCPJobTaxInput(BaseModel)`¶

Input to JobTaxMCPBlock — 26-op dispatch.

Field	Type	Default
`op`	`Literal['prepare_return', 'calculate_liability', 'file_declaration', 'review_deductions', 'process_refund', 'analyze_brackets', 'evaluate_exemptions', 'assess_compliance', 'benchmark_rates', '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', '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`	''

`MCPJobTaxOutput(BaseModel)`¶

Output from JobTaxMCPBlock.

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
`human_review_required`	`bool`	False
`suppression_cause`	`str \\| None`	None

`TaxStore(JobStore)`¶

SQLite store for the Tax job agent.

Constructor:

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

Methods:

`create_tax_return(taxpayer_id: str = '', tax_year: int = 2024, filing_status: str = 'single', jurisdiction: str = 'US', gross_income: str = '0', taxable_income: str = '0', tax_liability: str = '0', status: str = 'draft', data: dict | None = None) -> str`¶

`get_tax_return(return_id: str) -> dict | None`¶

`update_tax_return(return_id: str, **fields: Any) -> bool`¶

`list_tax_returns(taxpayer_id: str = '', status: str = '', limit: int = 50) -> list[dict]`¶

`add_deduction(return_id: str = '', category: str = '', description: str = '', amount: str = '0', substantiated: bool = False, data: dict | None = None) -> str`¶

`list_deductions(return_id: str = '', category: str = '', limit: int = 100) -> list[dict]`¶

`get_deductions_total(return_id: str) -> str`¶

`create_entity(name: str, entity_type: str = 'individual', ein: str = '', jurisdiction: str = 'US', fiscal_year_end: str = 'December 31', data: dict | None = None) -> str`¶

`get_entity(entity_id: str) -> dict | None`¶

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

`create_filing(entity_id: str = '', form_type: str = '', due_date: str = '', filed_date: str = '', status: str = 'pending', data: dict | None = None) -> str`¶

`update_filing(filing_id: str, **fields: Any) -> bool`¶

`list_filings(entity_id: str = '', status: str = '', limit: int = 50) -> list[dict]`¶

`create_plan(entity_id: str = '', strategy: str = '', estimated_savings: str = '0', risk_level: str = 'low', data: dict | None = None) -> str`¶

`list_plans(entity_id: str = '', risk_level: str = '', limit: int = 50) -> list[dict]`¶

`JobTaxMCPBlock(AIBlock[MCPJobTaxInput, MCPJobTaxOutput, dict])`¶

26-op MCP block for the Tax job agent.

Field	Type	Default
`name`	`str`	'job_tax_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: MCPJobTaxInput) -> Result[MCPJobTaxOutput]`¶

Functions¶

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

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

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

Run grounded four-valued QA over a tax output.

MCP Tools¶

Operation	Source
`prepare_return`	tax_mcp
`calculate_liability`	tax_mcp
`file_declaration`	tax_mcp
`review_deductions`	tax_mcp
`process_refund`	tax_mcp
`analyze_brackets`	tax_mcp
`evaluate_exemptions`	tax_mcp
`assess_compliance`	tax_mcp
`benchmark_rates`	tax_mcp
`audit_entries`	tax_mcp
`create_proposal`	tax_mcp
`review_deliverable`	tax_mcp
`delegate_task`	tax_mcp
`report_status`	tax_mcp
`request_feedback`	tax_mcp
`store_artifact`	tax_mcp
`retrieve_artifact`	tax_mcp
`list_artifacts`	tax_mcp
`search_artifacts`	tax_mcp
`archive`	tax_mcp
`plan_sprint`	tax_mcp
`track_progress`	tax_mcp
`reflect_on_outcome`	tax_mcp
`get_capabilities`	tax_mcp
`info`	tax_mcp
`list_patterns`	tax_mcp

Job Tax¶

Overview¶

Public API¶

JobTaxInput(JobInput)¶

JobTaxOutput(JobOutput)¶

JobTaxBlock(JobAgentBlock)¶

MCPJobTaxInput(BaseModel)¶

MCPJobTaxOutput(BaseModel)¶

TaxStore(JobStore)¶

create_tax_return(taxpayer_id: str = '', tax_year: int = 2024, filing_status: str = 'single', jurisdiction: str = 'US', gross_income: str = '0', taxable_income: str = '0', tax_liability: str = '0', status: str = 'draft', data: dict | None = None) -> str¶

get_tax_return(return_id: str) -> dict | None¶

update_tax_return(return_id: str, **fields: Any) -> bool¶

list_tax_returns(taxpayer_id: str = '', status: str = '', limit: int = 50) -> list[dict]¶

add_deduction(return_id: str = '', category: str = '', description: str = '', amount: str = '0', substantiated: bool = False, data: dict | None = None) -> str¶

list_deductions(return_id: str = '', category: str = '', limit: int = 100) -> list[dict]¶

get_deductions_total(return_id: str) -> str¶

create_entity(name: str, entity_type: str = 'individual', ein: str = '', jurisdiction: str = 'US', fiscal_year_end: str = 'December 31', data: dict | None = None) -> str¶

get_entity(entity_id: str) -> dict | None¶

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

create_filing(entity_id: str = '', form_type: str = '', due_date: str = '', filed_date: str = '', status: str = 'pending', data: dict | None = None) -> str¶

update_filing(filing_id: str, **fields: Any) -> bool¶

list_filings(entity_id: str = '', status: str = '', limit: int = 50) -> list[dict]¶

create_plan(entity_id: str = '', strategy: str = '', estimated_savings: str = '0', risk_level: str = 'low', data: dict | None = None) -> str¶

list_plans(entity_id: str = '', risk_level: str = '', limit: int = 50) -> list[dict]¶

JobTaxMCPBlock(AIBlock[MCPJobTaxInput, MCPJobTaxOutput, dict])¶

infer(data: MCPJobTaxInput) -> Result[MCPJobTaxOutput]¶