Job Sales¶

job_sales — G6 Sales job agent.

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

Overview¶

Domain-specialist job agent for sales professionals and account executives. Qualifies inbound leads, drafts tailored proposals, forecasts pipeline by stage, negotiates deal terms, tracks closed deals, and analyses funnel conversion metrics — providing a structured sales-process workflow within G6's safety-bounded, audit-trailed JobAgentBlock framework.

Launch readiness caveat

job_sales is functionally useful for local MCP pilots and design-partner workflows, and its handlers tolerate common non-developer input formats such as currency strings, percentages, and partially malformed CRM exports. The remaining caveat is test-feedback speed rather than a known runtime defect: the focused sales test suite currently takes about 2.5 minutes on the reference Windows workstation. Treat this as a CI/developer-experience warning before tightening warning-as-error or per-component gating, not as a blocker for first-user sales workflows.

When to use:

Qualifying a new lead against ICP criteria and generating a personalised outreach brief
Drafting a structured commercial proposal with pricing, scope, and terms
Forecasting quarterly pipeline revenue by probability-weighted stage
Analysing funnel conversion rates to identify the largest drop-off point

Example:

from mvp.job_sales import JobSalesBlock, JobSalesInput

block = JobSalesBlock()
result = block.infer(JobSalesInput(
    task="Qualify the inbound lead from GlobalRetail Ltd and draft a discovery-call brief with recommended qualification questions",
    context={"lead_company": "GlobalRetail Ltd", "annual_revenue_aud": 120_000_000, "pain_points": ["inventory visibility", "supplier onboarding"]},
))
# result.ok → True; result.value → JobSalesOutput with result, artifacts

Works well with: job_framework, job_marketing, job_business

Public API¶

`JobSalesBlock(JobAgentBlock)`¶

G6 Sales job agent — delegates to Sales MCP block.

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

Methods:

`infer(data)`¶

`JobSalesInput(JobInput)`¶

Input for the Sales job agent.

`JobSalesOutput(JobOutput)`¶

`JobSalesMCPBlock(AIBlock[MCPJobSalesInput, MCPJobSalesOutput, dict])`¶

26-op MCP block for the Sales job agent.

Field	Type	Default
`name`	`str`	'job_sales_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: MCPJobSalesInput) -> Result[MCPJobSalesOutput]`¶

`MCPJobSalesInput(BaseModel)`¶

Input to JobSalesMCPBlock — 26-op dispatch.

Field	Type	Default
`op`	`Literal['qualify_lead', 'draft_proposal', 'forecast_pipeline', 'negotiate_terms', 'close_deal', 'analyze_funnel', 'evaluate_territory', 'assess_opportunity', 'benchmark_quota', 'audit_crm', '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`	''

`MCPJobSalesOutput(BaseModel)`¶

Output from JobSalesMCPBlock.

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] \\| None`	None
`request_id`	`str`	''
`run_id`	`str`	''

`SalesStore(JobStore)`¶

SQLite store for the Sales job agent.

Constructor:

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

Methods:

`create_lead(name: str, company: str = '', source: str = '', score: float = 0, status: str = 'new', contact: dict | None = None, data: dict | None = None) -> str`¶

`get_lead(lead_id: str) -> dict | None`¶

`list_leads(status: str = '', limit: int = 100) -> list[dict]`¶

`update_lead(lead_id: str, **fields: Any) -> bool`¶

`create_opportunity(lead_id: str = '', title: str = '', value: float = 0, stage: str = 'prospecting', probability: float = 0, close_date: str = '', products: list | None = None, data: dict | None = None) -> str`¶

`get_opportunity(opp_id: str) -> dict | None`¶

`list_opportunities(stage: str = '', lead_id: str = '', limit: int = 100) -> list[dict]`¶

`update_opportunity(opp_id: str, **fields: Any) -> bool`¶

`log_activity(opp_id: str = '', activity_type: str = '', subject: str = '', outcome: str = '', next_steps: str = '', activity_date: str = '', data: dict | None = None) -> str`¶

`get_activities(opp_id: str = '', activity_type: str = '', limit: int = 100) -> list[dict]`¶

`create_quotation(opp_id: str = '', items: list | None = None, subtotal: float = 0, discount: float = 0, tax: float = 0, total: float = 0, valid_until: str = '', status: str = 'draft') -> str`¶

`get_quotation(quote_id: str) -> dict | None`¶

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

`save_pipeline_snapshot(period: str, stages: dict | None = None, total_value: float = 0, weighted_value: float = 0, win_rate: float = 0, data: dict | None = None) -> str`¶

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

Functions¶

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

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

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

Run grounded four-valued QA over a sales output.

MCP Tools¶

Operation	Source
`qualify_lead`	sales_mcp
`draft_proposal`	sales_mcp
`forecast_pipeline`	sales_mcp
`negotiate_terms`	sales_mcp
`close_deal`	sales_mcp
`analyze_funnel`	sales_mcp
`evaluate_territory`	sales_mcp
`assess_opportunity`	sales_mcp
`benchmark_quota`	sales_mcp
`audit_crm`	sales_mcp
`create_proposal`	sales_mcp
`review_deliverable`	sales_mcp
`delegate_task`	sales_mcp
`report_status`	sales_mcp
`request_feedback`	sales_mcp
`store_artifact`	sales_mcp
`retrieve_artifact`	sales_mcp
`list_artifacts`	sales_mcp
`search_artifacts`	sales_mcp
`archive`	sales_mcp
`plan_sprint`	sales_mcp
`track_progress`	sales_mcp
`reflect_on_outcome`	sales_mcp
`get_capabilities`	sales_mcp
`info`	sales_mcp
`list_patterns`	sales_mcp

Job Sales¶

Overview¶

Public API¶

JobSalesBlock(JobAgentBlock)¶

infer(data)¶

JobSalesInput(JobInput)¶

JobSalesOutput(JobOutput)¶

JobSalesMCPBlock(AIBlock[MCPJobSalesInput, MCPJobSalesOutput, dict])¶

infer(data: MCPJobSalesInput) -> Result[MCPJobSalesOutput]¶

MCPJobSalesInput(BaseModel)¶

MCPJobSalesOutput(BaseModel)¶

SalesStore(JobStore)¶

create_lead(name: str, company: str = '', source: str = '', score: float = 0, status: str = 'new', contact: dict | None = None, data: dict | None = None) -> str¶

get_lead(lead_id: str) -> dict | None¶

list_leads(status: str = '', limit: int = 100) -> list[dict]¶

update_lead(lead_id: str, **fields: Any) -> bool¶

create_opportunity(lead_id: str = '', title: str = '', value: float = 0, stage: str = 'prospecting', probability: float = 0, close_date: str = '', products: list | None = None, data: dict | None = None) -> str¶

get_opportunity(opp_id: str) -> dict | None¶

list_opportunities(stage: str = '', lead_id: str = '', limit: int = 100) -> list[dict]¶

update_opportunity(opp_id: str, **fields: Any) -> bool¶

log_activity(opp_id: str = '', activity_type: str = '', subject: str = '', outcome: str = '', next_steps: str = '', activity_date: str = '', data: dict | None = None) -> str¶

get_activities(opp_id: str = '', activity_type: str = '', limit: int = 100) -> list[dict]¶

create_quotation(opp_id: str = '', items: list | None = None, subtotal: float = 0, discount: float = 0, tax: float = 0, total: float = 0, valid_until: str = '', status: str = 'draft') -> str¶

get_quotation(quote_id: str) -> dict | None¶

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

save_pipeline_snapshot(period: str, stages: dict | None = None, total_value: float = 0, weighted_value: float = 0, win_rate: float = 0, data: dict | None = None) -> str¶

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

Functions¶

assemble_review_text(output: Any) -> str¶

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

MCP Tools¶

`JobSalesBlock(JobAgentBlock)`¶

`infer(data)`¶

`JobSalesInput(JobInput)`¶

`JobSalesOutput(JobOutput)`¶

`JobSalesMCPBlock(AIBlock[MCPJobSalesInput, MCPJobSalesOutput, dict])`¶

`infer(data: MCPJobSalesInput) -> Result[MCPJobSalesOutput]`¶

`MCPJobSalesInput(BaseModel)`¶

`MCPJobSalesOutput(BaseModel)`¶

`SalesStore(JobStore)`¶

`create_lead(name: str, company: str = '', source: str = '', score: float = 0, status: str = 'new', contact: dict | None = None, data: dict | None = None) -> str`¶

`get_lead(lead_id: str) -> dict | None`¶

`list_leads(status: str = '', limit: int = 100) -> list[dict]`¶

`update_lead(lead_id: str, **fields: Any) -> bool`¶

`create_opportunity(lead_id: str = '', title: str = '', value: float = 0, stage: str = 'prospecting', probability: float = 0, close_date: str = '', products: list | None = None, data: dict | None = None) -> str`¶

`get_opportunity(opp_id: str) -> dict | None`¶

`list_opportunities(stage: str = '', lead_id: str = '', limit: int = 100) -> list[dict]`¶

`update_opportunity(opp_id: str, **fields: Any) -> bool`¶

`log_activity(opp_id: str = '', activity_type: str = '', subject: str = '', outcome: str = '', next_steps: str = '', activity_date: str = '', data: dict | None = None) -> str`¶

`get_activities(opp_id: str = '', activity_type: str = '', limit: int = 100) -> list[dict]`¶

`create_quotation(opp_id: str = '', items: list | None = None, subtotal: float = 0, discount: float = 0, tax: float = 0, total: float = 0, valid_until: str = '', status: str = 'draft') -> str`¶

`get_quotation(quote_id: str) -> dict | None`¶

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

`save_pipeline_snapshot(period: str, stages: dict | None = None, total_value: float = 0, weighted_value: float = 0, win_rate: float = 0, data: dict | None = None) -> str`¶

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

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

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