Job Marketing¶

job_marketing — G6 Marketing job agent.

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

Overview¶

Domain-specialist job agent for marketing professionals. Creates multi-channel campaigns, analyses market and competitor landscapes, segments target audiences, drafts copy, measures ROI against spend, and runs competitive benchmarking — providing a full marketing workflow within G6's safety-bounded, audit-trailed JobAgentBlock framework.

When to use:

• Designing a multi-channel campaign from brief through creative, targeting, and budget allocation
• Segmenting a customer database to identify the highest-value addressable audience
• Measuring campaign ROI and attributing performance across channels
• Conducting competitive analysis to surface positioning gaps and market opportunities

Example:

from mvp.job_marketing import JobMarketingBlock, JobMarketingInput

block = JobMarketingBlock()
result = block.infer(JobMarketingInput(
    task="Create a 6-week digital campaign plan to drive sign-ups for a B2B SaaS trial in the Australian mid-market",
    context={"budget_aud": 80_000, "channels": ["LinkedIn", "Google Ads", "email"], "target_signups": 500},
))
# result.ok → True; result.value → JobMarketingOutput with result, artifacts

Production metrics import:

measure_roi can read real campaign, website, YouTube, or social exports from CSV/JSON via connector="file":

result = block.infer(JobMarketingInput(
    task="Measure ROI for the launch campaign",
    parameters={
        "connector": "file",
        "metrics_path": "exports/launch_metrics.csv",
        "campaign_id": "launch",
    },
))

Accepted column aliases include campaign/campaign_id, channel/source/platform, impressions/views/sessions, clicks/link_clicks, conversions/signups/trials, spend/cost/amount_spent, and revenue/conversion_value. Real imports fail loudly for missing files, unmapped schemas, empty campaign matches, or missing positive spend, and include validation metadata showing which columns were mapped.

Reliability envelope:

Outputs expose completion_state, warning_card, and evidence metadata. Legal completion states are exactly verified, qualified-draft, and blocked-escalated. Offline agentic suppression and connector-unavailable paths return degraded but useful qualified-draft results with a warning card instead of silently overclaiming completion. get_capabilities reports marketing-specific integration probes for corpus grounding, ctx_search, adapt_experta, adapt_bayesian, debate, adapt_sklearn, and the mock/file connectors.

Works well with: job_framework, job_creative_media, job_sales

Public API¶

JobMarketingBlock(JobAgentBlock)¶

G6 Marketing job agent — delegates to Marketing MCP block.

Methods:

infer(data)¶

JobMarketingInput(JobInput)¶

Input for the Marketing job agent.

JobMarketingOutput(JobOutput)¶

JobMarketingMCPBlock(AIBlock[MCPJobMarketingInput, MCPJobMarketingOutput, dict])¶

26-op MCP block for the Marketing job agent.

Methods:

infer(data: MCPJobMarketingInput) -> Result[MCPJobMarketingOutput]¶

MCPJobMarketingInput(BaseModel)¶

Input to JobMarketingMCPBlock — 26-op dispatch.

MCPJobMarketingOutput(BaseModel)¶

Output from JobMarketingMCPBlock.

MarketingStore(JobStore)¶

SQLite store for the Marketing job agent.

Constructor:

Methods:

save_campaign(campaign_id: str = '', name: str = '', channel: str = 'multi-channel', objective: str = 'awareness', budget: float = 0, start_date: str = '', end_date: str = '', metrics: dict | None = None, status: str = 'draft') -> str¶

Insert or replace a campaign. Returns the campaign_id.

get_campaign(campaign_id: str) -> dict | None¶

Fetch a campaign by ID.

list_campaigns(status: str = '', channel: str = '') -> list[dict]¶

List campaigns, optionally filtering by status or channel.

update_campaign_status(campaign_id: str, status: str) -> bool¶

Update campaign status. Returns True if row was updated.

update_campaign_metrics(campaign_id: str, metrics: dict) -> bool¶

Merge new metrics into campaign metrics_json.

get_campaign_budget_total(status: str = '') -> float¶

Sum of budgets across campaigns, optionally filtered by status.

save_audience(audience_id: str = '', name: str = '', segment_criteria: dict | None = None, size_estimate: int = 0, demographics: dict | None = None, data: dict | None = None) -> str¶

Insert or replace an audience segment. Returns the audience_id.

get_audience(audience_id: str) -> dict | None¶

Fetch an audience by ID.

list_audiences(limit: int = 50) -> list[dict]¶

List all audience segments.

search_audiences(query: str, limit: int = 20) -> list[dict]¶

Search audiences by name or criteria.

save_content(content_id: str = '', campaign_id: str = '', content_type: str = 'copy', title: str = '', body: str = '', platform: str = 'general', performance: dict | None = None, status: str = 'draft') -> str¶

Insert or replace a content piece. Returns the content_id.

get_content(content_id: str) -> dict | None¶

Fetch a content piece by ID.

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

List content, optionally filtering by campaign or status.

update_content_status(content_id: str, status: str) -> bool¶

Update content status (draft -> review -> published -> archived).

update_content_performance(content_id: str, performance: dict) -> bool¶

Merge performance metrics into content performance_json.

save_analytics(analytics_id: str = '', campaign_id: str = '', period: str = '', impressions: int = 0, clicks: int = 0, conversions: int = 0, spend: float = 0, revenue: float = 0, data: dict | None = None) -> str¶

Insert an analytics record. Returns the analytics_id.

get_analytics(campaign_id: str = '', period: str = '', limit: int = 100) -> list[dict]¶

Retrieve analytics records with optional filters.

get_analytics_summary(campaign_id: str = '') -> dict¶

Aggregate analytics metrics across records for a campaign or all.

save_brand_asset(asset_id: str = '', name: str = '', asset_type: str = 'logo', guidelines: dict | None = None, usage_rules: str = '', version: str = '1.0', status: str = 'active') -> str¶

Insert or replace a brand asset. Returns the asset_id.

get_brand_asset(asset_id: str) -> dict | None¶

Fetch a brand asset by ID.

list_brand_assets(asset_type: str = '', status: str = 'active', limit: int = 50) -> list[dict]¶

List brand assets with optional filters.

save_work(work_id: str = '', category: str = '', title: str = '', data: dict | None = None, status: str = 'active') -> str¶

Insert or replace a marketing work item. Returns the id.

get_work(work_id: str) -> dict | None¶

Fetch a marketing work item by ID.

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

List marketing work items.

get_campaign_with_content(campaign_id: str) -> dict¶

Fetch a campaign with all its associated content pieces.

count_domain_tables() -> dict[str, int]¶

Count rows in all marketing-specific tables.

Functions¶

assemble_review_text(output: Any) -> str¶

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

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

Run grounded four-valued QA over a marketing output.

MCP Tools¶