Skip to content

Job Creative Media

job_creative_media — G6 CreativeMedia job agent.

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

Overview

Domain-specialist job agent for creative media professionals including writers, designers, and content strategists. It supports concept development, content and brand audits, copy review, trend research, campaign pitching, audience analysis, engagement scoring, and creative workflow tracking - all within G6's safety-bounded, audit-trailed JobAgentBlock framework.

Advisory and analytics scope

job_creative_media does not generate final creative copy, images, video, or production-ready media assets. Use it to analyze, score, plan, review, and structure creative work. Pair it with specialist generation tools or a human creative workflow when you need final copy, artwork, video, or production files.

Live web search and live LLM debate are disabled by default for first-run reliability. Set G6_JOB_LIVE_SEARCH=1 to allow live DuckDuckGo trend lookups and G6_JOB_LIVE_LLM=1 to allow live debate/enrichment; otherwise the component uses deterministic local-safe fallbacks.

The DAM adapter status is in_memory: InMemoryDAMAdapter is available for development and tests only, with no external Bynder/Canto/Cloudinary-style backend wired. MCP get_capabilities discloses this as development_only and simulation_only.

Grounded QA is available as the standalone callable assess_creative_media_output, but it is not automatic in the main execution path. Orchestrators that need post-hoc quality review must opt in explicitly.

Creative media is not registered in the framework HITL policy table. SENSITIVE_OPS (analyze_audience, audit_content, review_edit) set advisory human-review expectations at Tier-1, but MCP HITL checks do not block and report framework_policy: not_registered.

Reliability envelopes use the exact fields completion_state, warning_card, evidence, request_id, task_id, and run_id. Legal completion_state values are only verified, qualified-draft, and blocked-escalated.

When to use:

  • Creating structured content briefs, campaign outlines, and review checklists
  • Reviewing and editing existing content for tone, brand alignment, and clarity
  • Researching emerging trends to brief a creative team before a campaign kick-off
  • Pitching campaign concepts with supporting rationale and target audience data

Example:

from mvp.job_creative_media import JobCreativeMediaBlock, JobCreativeMediaInput

block = JobCreativeMediaBlock()
result = block.infer(JobCreativeMediaInput(
    task="Develop three distinct creative concepts for a product launch campaign targeting Gen Z consumers",
    context={"product": "EcoBottle Pro", "channels": ["TikTok", "Instagram", "YouTube Shorts"]},
))
# result.ok -> True; result.value -> JobCreativeMediaOutput with result, artifacts

Works well with: job_framework, job_marketing, job_public_relations

Public API

JobCreativeMediaBlock(JobAgentBlock)

G6 CreativeMedia job agent - Tier 1 block with MCP delegation.

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

Methods:

infer(data: Any)

JobCreativeMediaInput(JobInput)

Input for the CreativeMedia job agent.

JobCreativeMediaOutput(JobOutput)

DAMAdapter(ABC)

Abstract base class for Digital Asset Management integration.

Methods:

upload_asset(file_path: str, metadata: dict[str, Any] | None = None) -> str

Upload an asset and return its ID.

download_asset(asset_id: str) -> dict[str, Any]

Download/retrieve an asset by ID.

search_assets(query: str, filters: dict[str, Any] | None = None, limit: int = 20) -> list[dict[str, Any]]

Search assets by query string and optional filters.

update_metadata(asset_id: str, metadata: dict[str, Any]) -> bool

Update metadata for an existing asset.

delete_asset(asset_id: str) -> bool

Delete an asset by ID.

list_versions(asset_id: str) -> list[dict[str, Any]]

List all versions of an asset.

create_collection(name: str, description: str = '') -> str

Create a named collection/folder and return its ID.

add_to_collection(collection_id: str, asset_id: str) -> bool

Add an asset to a collection.

InMemoryDAMAdapter(DAMAdapter)

In-memory DAM adapter for testing and development.

Methods:

upload_asset(file_path: str, metadata: dict[str, Any] | None = None) -> str

download_asset(asset_id: str) -> dict[str, Any]

search_assets(query: str, filters: dict[str, Any] | None = None, limit: int = 20) -> list[dict[str, Any]]

update_metadata(asset_id: str, metadata: dict[str, Any]) -> bool

delete_asset(asset_id: str) -> bool

list_versions(asset_id: str) -> list[dict[str, Any]]

upload_new_version(asset_id: str, file_path: str) -> int

Upload a new version of an existing asset. Returns version number.

create_collection(name: str, description: str = '') -> str

add_to_collection(collection_id: str, asset_id: str) -> bool

list_collection(collection_id: str) -> dict[str, Any]

List contents of a collection.

JobCreativeMediaMCPBlock(AIBlock[MCPJobCreativeMediaInput, MCPJobCreativeMediaOutput, dict])

26-op MCP block for the CreativeMedia job agent.

Field Type Default
name str 'job_creative_media_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: MCPJobCreativeMediaInput) -> Result[MCPJobCreativeMediaOutput]

MCPJobCreativeMediaInput(BaseModel)

Input to JobCreativeMediaMCPBlock - 26-op dispatch.

Field Type Default
op Literal['develop_concept', 'draft_content', 'review_edit', 'research_trend', 'pitch_idea', 'analyze_audience', 'evaluate_engagement', 'assess_impact', 'benchmark_reach', 'audit_content', '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 ''

MCPJobCreativeMediaOutput(BaseModel)

Output from JobCreativeMediaMCPBlock.

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] 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

CreativeMediaStore(JobStore)

SQLite store for the CreativeMedia job agent.

Constructor:

Parameter Type Default
db_path str ':memory:'

Methods:

create_project(title: str = '', project_type: str = '', client: str = '', brief: dict | None = None, deliverables: list | None = None, deadline: str = '', budget: float = 0.0) -> str

get_project(project_id: str) -> dict | None

list_projects(status: str = '', client: str = '', limit: int = 50) -> list[dict]

update_project(project_id: str, **fields: Any) -> bool

create_asset(project_id: str = '', asset_type: str = '', title: str = '', fmt: str = '', dimensions: dict | None = None, file_path: str = '', metadata: dict | None = None, version: int = 1) -> str

get_asset(asset_id: str) -> dict | None

list_assets(project_id: str = '', asset_type: str = '', limit: int = 50) -> list[dict]

update_asset(asset_id: str, **fields: Any) -> bool

create_brand_guidelines(client: str = '', colors: list | None = None, typography: dict | None = None, logo_usage: dict | None = None, tone_voice: str = '', guidelines: dict | None = None, version: int = 1) -> str

get_brand_guidelines(brand_id: str) -> dict | None

list_brand_guidelines(client: str = '', limit: int = 50) -> list[dict]

create_calendar_entry(project_id: str = '', platform: str = '', content_type: str = '', scheduled_date: str = '', copy: str = '', hashtags: list | None = None, metrics: dict | None = None) -> str

list_calendar(project_id: str = '', platform: str = '', limit: int = 50) -> list[dict]

update_calendar_entry(entry_id: str, **fields: Any) -> bool

create_review(asset_id: str = '', reviewer: str = '', feedback: dict | None = None, revision_round: int = 1, approval_status: str = 'pending') -> str

list_reviews(asset_id: str = '', approval_status: str = '', limit: int = 50) -> list[dict]

update_review(review_id: str, **fields: Any) -> bool

Functions

assemble_review_text(output: Any) -> str

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

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

Run grounded four-valued QA over a creative-media output.

run_creative_brief_to_delivery(store: CreativeMediaStore, parameters: dict[str, Any] | None = None) -> dict[str, Any]

Execute the full creative workflow: brief -> concept -> production -> review -> delivery.

MCP Tools

Operation Source
develop_concept creative_media_mcp
draft_content creative_media_mcp
review_edit creative_media_mcp
research_trend creative_media_mcp
pitch_idea creative_media_mcp
analyze_audience creative_media_mcp
evaluate_engagement creative_media_mcp
assess_impact creative_media_mcp
benchmark_reach creative_media_mcp
audit_content creative_media_mcp
create_proposal creative_media_mcp
review_deliverable creative_media_mcp
delegate_task creative_media_mcp
report_status creative_media_mcp
request_feedback creative_media_mcp
store_artifact creative_media_mcp
retrieve_artifact creative_media_mcp
list_artifacts creative_media_mcp
search_artifacts creative_media_mcp
archive creative_media_mcp
plan_sprint creative_media_mcp
track_progress creative_media_mcp
reflect_on_outcome creative_media_mcp
list_patterns creative_media_mcp
get_capabilities creative_media_mcp
info creative_media_mcp