Job Business¶

job_business — G6 Business job agent.

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

Overview¶

Domain-specialist job agent for general business development and strategy. Evaluates market opportunities, develops strategic roadmaps, assesses partnerships, builds business cases, monitors KPIs, and conducts competitive market analysis — delivering structured strategic outputs within G6's safety-bounded, audit-trailed JobAgentBlock framework.

When to use:

Evaluating a new market entry opportunity with a structured business case and risk summary
Developing a strategic roadmap aligned to organisational OKRs and resource constraints
Assessing potential partnership or acquisition targets against defined criteria
Monitoring business KPIs and alerting on variance from targets

Caveats for production use:

Live web search is disabled by default. Set G6_JOB_LIVE_SEARCH=1 only when the deployment has approved network access and search latency/costs are acceptable.
Live LLM debate/enrichment is disabled by default. Set G6_JOB_LIVE_LLM=1 only when a configured LLM backend is healthy; otherwise the agent uses deterministic/mock debate so first-run MCP workflows remain fast and offline-safe.
Business outputs are decision-support artifacts, not legal, tax, investment, accounting, or regulated professional advice. Review assumptions and numbers before using them for customer-facing or board-level decisions.
Calculations depend on user-provided inputs such as TAM assumptions, CAC, runway, current liabilities, and market share. Missing or placeholder inputs produce illustrative outputs rather than validated forecasts.
The default external-data connector is mock; mock-backed market, KPI, and competitor data is surfaced as completion_state: qualified-draft with warning_card.code: mock_connector.
Optional integration failures are surfaced as completion_state: qualified-draft with warning_card.code: integration_degraded when deterministic business rules can still return useful output.
grounded_qa.assess_business_output is available as an explicit post-hoc QA utility. It is not part of the default dispatch path.

Example:

from mvp.job_business import JobBusinessBlock, JobBusinessInput

block = JobBusinessBlock()
result = block.infer(JobBusinessInput(
    task="Evaluate the opportunity to expand into the Southeast Asian SaaS market by Q3 2026",
    context={"budget_usd": 500_000, "headcount_available": 5},
))
# result.ok → True; result.value → JobBusinessOutput with result, artifacts

Works well with: job_framework, job_consultant, job_analyst

Public API¶

`JobBusinessBlock(JobAgentBlock)`¶

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

Methods:

`infer(data)`¶

`JobBusinessInput(JobInput)`¶

Input for the Business job agent.

`JobBusinessOutput(JobOutput)`¶

Output from the Business job agent.

`JobBusinessMCPBlock(AIBlock[MCPJobBusinessInput, MCPJobBusinessOutput, dict])`¶

26-op MCP block for the Business job agent.

Field	Type	Default
`name`	`str`	'job_business_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: MCPJobBusinessInput) -> Result[MCPJobBusinessOutput]`¶

`MCPJobBusinessInput(BaseModel)`¶

Input to JobBusinessMCPBlock - 26-op dispatch.

Field	Type	Default
`op`	`Literal['analyze_opportunity', 'develop_strategy', 'evaluate_partnership', 'create_business_case', 'monitor_kpis', 'analyze_market', 'evaluate_competition', 'assess_viability', 'benchmark_growth', 'audit_financials', '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']`	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`	''

`MCPJobBusinessOutput(BaseModel)`¶

Output from JobBusinessMCPBlock.

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`	''
`human_review_required`	`bool`	False
`review_status`	`ReviewStatus \\| None`	None

`BusinessStore(JobStore)`¶

SQLite store for the Business job agent.

Constructor:

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

Methods:

`create_business_plan(company_name: str, industry: str = '', stage: str = 'idea', revenue_model: str = '', target_market: str = '', value_proposition: str = '', data: dict | None = None) -> str`¶

Insert a new business plan. Returns plan_id.

`get_business_plan(plan_id: str) -> dict | None`¶

Fetch a single business plan by ID.

`update_business_plan(plan_id: str, **fields: Any) -> bool`¶

Update mutable fields on a business plan.

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

List business plans with optional filters.

`delete_business_plan(plan_id: str) -> bool`¶

Delete a business plan by ID.

`create_market_analysis(plan_id: str = '', industry: str = '', market_size: float = 0.0, tam: float = 0.0, sam: float = 0.0, som: float = 0.0, growth_rate: float = 0.0, concentration_hhi: float = 0.0, segments: list | None = None, findings: dict | None = None) -> str`¶

Insert a new market analysis. Returns analysis_id.

`get_market_analysis(analysis_id: str) -> dict | None`¶

Fetch a single market analysis by ID.

`list_market_analyses(plan_id: str = '', industry: str = '', limit: int = 50) -> list[dict]`¶

List market analyses with optional filters.

`delete_market_analysis(analysis_id: str) -> bool`¶

Delete a market analysis by ID.

`create_competitive_intel(plan_id: str = '', competitor_name: str = '', market_share: float = 0.0, strengths: list | None = None, weaknesses: list | None = None, scores: dict | None = None, positioning: str = '', threat_level: str = 'medium', data: dict | None = None) -> str`¶

Insert a new competitive intelligence entry. Returns intel_id.

`get_competitive_intel(intel_id: str) -> dict | None`¶

Fetch a single competitive intel entry by ID.

`list_competitive_intel(plan_id: str = '', threat_level: str = '', limit: int = 50) -> list[dict]`¶

List competitive intel entries with optional filters.

`update_competitive_intel(intel_id: str, **fields: Any) -> bool`¶

Update mutable fields on a competitive intel entry.

`delete_competitive_intel(intel_id: str) -> bool`¶

Delete a competitive intel entry by ID.

`create_financial_projection(plan_id: str = '', period: str = '', revenue: float = 0.0, costs: float = 0.0, profit: float = 0.0, gross_margin_pct: float = 0.0, burn_rate: float = 0.0, runway_months: float = 0.0, assumptions: dict | None = None) -> str`¶

Insert a new financial projection. Returns projection_id.

`get_financial_projection(projection_id: str) -> dict | None`¶

Fetch a single financial projection by ID.

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

List financial projections for a plan.

`delete_financial_projection(projection_id: str) -> bool`¶

Delete a financial projection by ID.

`create_operational_metric(department: str = '', metric_name: str = '', value: float = 0.0, target: float = 0.0, period: str = '', data: dict | None = None) -> str`¶

Insert a new operational metric. Returns metric_id.

`list_operational_metrics(department: str = '', period: str = '', limit: int = 50) -> list[dict]`¶

List operational metrics with optional filters.

`create_strategic_initiative(title: str, category: str = '', priority: str = 'medium', budget: float = 0.0, roi_estimate: float = 0.0, data: dict | None = None) -> str`¶

Insert a new strategic initiative. Returns initiative_id.

`get_strategic_initiative(initiative_id: str) -> dict | None`¶

Fetch a single strategic initiative by ID.

`update_strategic_initiative(initiative_id: str, **fields: Any) -> bool`¶

Update mutable fields on a strategic initiative.

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

List strategic initiatives with optional filters.

Functions¶

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

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

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

Run grounded four-valued QA over a business output.

MCP Tools¶

Operation	Source
`verified`	business_mcp
`qualified-draft`	business_mcp
`blocked-escalated`	business_mcp
`not_required`	business_mcp
`pending`	business_mcp
`approved`	business_mcp
`bypassed`	business_mcp
`analyze_opportunity`	business_mcp
`develop_strategy`	business_mcp
`evaluate_partnership`	business_mcp
`create_business_case`	business_mcp
`monitor_kpis`	business_mcp
`analyze_market`	business_mcp
`evaluate_competition`	business_mcp
`assess_viability`	business_mcp
`benchmark_growth`	business_mcp
`audit_financials`	business_mcp
`create_proposal`	business_mcp
`review_deliverable`	business_mcp
`delegate_task`	business_mcp
`report_status`	business_mcp
`request_feedback`	business_mcp
`store_artifact`	business_mcp
`retrieve_artifact`	business_mcp
`list_artifacts`	business_mcp
`search_artifacts`	business_mcp
`archive`	business_mcp
`plan_sprint`	business_mcp
`track_progress`	business_mcp
`reflect_on_outcome`	business_mcp
`list_patterns`	business_mcp
`get_capabilities`	business_mcp
`info`	business_mcp

Job Business¶

Overview¶

Public API¶

JobBusinessBlock(JobAgentBlock)¶

infer(data)¶

JobBusinessInput(JobInput)¶

JobBusinessOutput(JobOutput)¶

JobBusinessMCPBlock(AIBlock[MCPJobBusinessInput, MCPJobBusinessOutput, dict])¶

infer(data: MCPJobBusinessInput) -> Result[MCPJobBusinessOutput]¶

MCPJobBusinessInput(BaseModel)¶

MCPJobBusinessOutput(BaseModel)¶

BusinessStore(JobStore)¶

create_business_plan(company_name: str, industry: str = '', stage: str = 'idea', revenue_model: str = '', target_market: str = '', value_proposition: str = '', data: dict | None = None) -> str¶

get_business_plan(plan_id: str) -> dict | None¶

update_business_plan(plan_id: str, **fields: Any) -> bool¶

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

delete_business_plan(plan_id: str) -> bool¶

create_market_analysis(plan_id: str = '', industry: str = '', market_size: float = 0.0, tam: float = 0.0, sam: float = 0.0, som: float = 0.0, growth_rate: float = 0.0, concentration_hhi: float = 0.0, segments: list | None = None, findings: dict | None = None) -> str¶

get_market_analysis(analysis_id: str) -> dict | None¶

list_market_analyses(plan_id: str = '', industry: str = '', limit: int = 50) -> list[dict]¶

delete_market_analysis(analysis_id: str) -> bool¶

create_competitive_intel(plan_id: str = '', competitor_name: str = '', market_share: float = 0.0, strengths: list | None = None, weaknesses: list | None = None, scores: dict | None = None, positioning: str = '', threat_level: str = 'medium', data: dict | None = None) -> str¶

get_competitive_intel(intel_id: str) -> dict | None¶

list_competitive_intel(plan_id: str = '', threat_level: str = '', limit: int = 50) -> list[dict]¶

update_competitive_intel(intel_id: str, **fields: Any) -> bool¶

delete_competitive_intel(intel_id: str) -> bool¶

create_financial_projection(plan_id: str = '', period: str = '', revenue: float = 0.0, costs: float = 0.0, profit: float = 0.0, gross_margin_pct: float = 0.0, burn_rate: float = 0.0, runway_months: float = 0.0, assumptions: dict | None = None) -> str¶

get_financial_projection(projection_id: str) -> dict | None¶

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

delete_financial_projection(projection_id: str) -> bool¶

create_operational_metric(department: str = '', metric_name: str = '', value: float = 0.0, target: float = 0.0, period: str = '', data: dict | None = None) -> str¶

list_operational_metrics(department: str = '', period: str = '', limit: int = 50) -> list[dict]¶

create_strategic_initiative(title: str, category: str = '', priority: str = 'medium', budget: float = 0.0, roi_estimate: float = 0.0, data: dict | None = None) -> str¶

get_strategic_initiative(initiative_id: str) -> dict | None¶

update_strategic_initiative(initiative_id: str, **fields: Any) -> bool¶

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

Functions¶

assemble_review_text(output: Any) -> str¶

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

MCP Tools¶

`JobBusinessBlock(JobAgentBlock)`¶

`infer(data)`¶

`JobBusinessInput(JobInput)`¶

`JobBusinessOutput(JobOutput)`¶

`JobBusinessMCPBlock(AIBlock[MCPJobBusinessInput, MCPJobBusinessOutput, dict])`¶

`infer(data: MCPJobBusinessInput) -> Result[MCPJobBusinessOutput]`¶

`MCPJobBusinessInput(BaseModel)`¶

`MCPJobBusinessOutput(BaseModel)`¶

`BusinessStore(JobStore)`¶

`create_business_plan(company_name: str, industry: str = '', stage: str = 'idea', revenue_model: str = '', target_market: str = '', value_proposition: str = '', data: dict | None = None) -> str`¶

`get_business_plan(plan_id: str) -> dict | None`¶

`update_business_plan(plan_id: str, **fields: Any) -> bool`¶

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

`delete_business_plan(plan_id: str) -> bool`¶

`create_market_analysis(plan_id: str = '', industry: str = '', market_size: float = 0.0, tam: float = 0.0, sam: float = 0.0, som: float = 0.0, growth_rate: float = 0.0, concentration_hhi: float = 0.0, segments: list | None = None, findings: dict | None = None) -> str`¶

`get_market_analysis(analysis_id: str) -> dict | None`¶

`list_market_analyses(plan_id: str = '', industry: str = '', limit: int = 50) -> list[dict]`¶

`delete_market_analysis(analysis_id: str) -> bool`¶

`create_competitive_intel(plan_id: str = '', competitor_name: str = '', market_share: float = 0.0, strengths: list | None = None, weaknesses: list | None = None, scores: dict | None = None, positioning: str = '', threat_level: str = 'medium', data: dict | None = None) -> str`¶

`get_competitive_intel(intel_id: str) -> dict | None`¶

`list_competitive_intel(plan_id: str = '', threat_level: str = '', limit: int = 50) -> list[dict]`¶

`update_competitive_intel(intel_id: str, **fields: Any) -> bool`¶

`delete_competitive_intel(intel_id: str) -> bool`¶

`create_financial_projection(plan_id: str = '', period: str = '', revenue: float = 0.0, costs: float = 0.0, profit: float = 0.0, gross_margin_pct: float = 0.0, burn_rate: float = 0.0, runway_months: float = 0.0, assumptions: dict | None = None) -> str`¶

`get_financial_projection(projection_id: str) -> dict | None`¶

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

`delete_financial_projection(projection_id: str) -> bool`¶

`create_operational_metric(department: str = '', metric_name: str = '', value: float = 0.0, target: float = 0.0, period: str = '', data: dict | None = None) -> str`¶

`list_operational_metrics(department: str = '', period: str = '', limit: int = 50) -> list[dict]`¶

`create_strategic_initiative(title: str, category: str = '', priority: str = 'medium', budget: float = 0.0, roi_estimate: float = 0.0, data: dict | None = None) -> str`¶

`get_strategic_initiative(initiative_id: str) -> dict | None`¶

`update_strategic_initiative(initiative_id: str, **fields: Any) -> bool`¶

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

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

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