from framework.agents.agent import Agent
from framework.decorators import agent


@agent(
    name="RevenueAgent",
    label="Revenue",
    description="Analyzes pipeline, accounts, and forecast risk",
    role="You are a revenue operations specialist.",
    blocks=["role", "communication", "tool_guidance", "error_handling"],
    llm_config={"model": "openai/gpt-5", "temperature": 0.2},
    reasoning_effort="medium",
    max_tokens=4000,
    enable_planning=True,
    enable_reflection=True,
    icon="chart-bar",
)
class RevenueAgent(Agent):
    pass

from framework.agents.sub_agent import SubAgent
from framework.decorators import sub_agent
from framework.models.approval_models import ApprovalConfig


@sub_agent(
    name="ContractReviewAgent",
    label="Contract review",
    description="Reviews contract terms and prepares risk summaries",
    parents=["RevenueAgent", "MainAgent"],
    role="You review commercial contracts for revenue and renewal risk.",
    blocks=["role", "sub_agent_context", "communication", "tool_guidance", "citations"],
    llm_config={"model": "openai/gpt-5", "temperature": 0.1},
    reasoning_effort="high",
    require_approval=ApprovalConfig(enabled=True, timeout_seconds=600),
)
class ContractReviewAgent(SubAgent):
    pass

@sub_agent(
    name="BillingAgent",
    parents=["MainAgent"],
    require_approval={
        "enabled": True,
        "timeout_seconds": 900,
        "auto_approve_after_success": 0,
        "escalate_on_timeout": True,
    },
)
class BillingAgent(SubAgent):
    pass

from framework.decorators import tool
from framework.models.approval_models import ApprovalConfig


@tool(
    name="create_renewal_task",
    label="Create renewal task",
    description="Create a follow-up task for a renewal opportunity",
    agent=["RevenueAgent", "ContractReviewAgent"],
    tags=["tasks", "renewals"],
    group="Revenue Operations",
    require_approval=ApprovalConfig(enabled=True, timeout_seconds=300),
    source_type="tasks",
    show_in_timeline=True,
)
async def create_renewal_task(
    account_id: str,
    owner_id: str,
    due_date: str,
    note: str,
) -> dict:
    """Create a renewal follow-up task.

    Args:
        account_id: CRM account identifier.
        owner_id: User who owns the task.
        due_date: ISO date for the task.
        note: Task body.
    """
    ...

from collections.abc import AsyncGenerator
from typing import Any

from framework.decorators import tool
from framework.models.executable import Event, EventType


@tool(agent="MainAgent", show_prompt_to_user=True)
async def request_missing_fields(fields: list[str], **kwargs) -> AsyncGenerator[Any, None]:
    call_id = kwargs["call_id"]

    yield Event(
        type=EventType.QUESTION_REQUIRED,
        call_id=call_id,
        content={"fields": fields},
    )

    answer = await wait_for_user_answer(call_id)
    yield {"fields": answer}

from pydantic import BaseModel, Field

from framework.artifacts.action_specs import EventAction, StateAction
from framework.artifacts.specs import artifact
from framework.tools.result_summarizers import register_cached_tool


class AccountHealthContent(BaseModel):
    account_id: str
    account_name: str
    health_score: int
    risks: list[str] = Field(default_factory=list)


@artifact(
    artifact_type="account_health",
    label="Account Health",
    icon="heart-pulse",
    description="Account health summary with risks and recommendations",
    display_mode="panel",
    supports_streaming=True,
    agent=["RevenueAgent"],
    content_schema=AccountHealthContent,
    actions={
        "acknowledged": StateAction(timestamp_key="acknowledged_at"),
        "note_added": EventAction(),
    },
)
def summarize_account_health(content: dict) -> dict:
    return {
        "title": content.get("account_name", "Account health"),
        "score": content.get("health_score"),
        "risk_count": len(content.get("risks") or []),
    }


register_cached_tool(
    tool_name="get_account_health",
    summarizer=summarize_account_health,
)

---
agents: ["RevenueAgent"]
category: "capabilities"
description: "How to handle renewal workflows"
---

When working on renewals, check open opportunities, account health, and recent emails before recommending next steps.

from framework.prompts.block_registry import PromptContext, prompt_block


@prompt_block(
    scope="conversation",
    tags=["renewals", "context"],
    ttl=300,
    agent=["RevenueAgent"],
)
async def renewal_context(ctx: PromptContext) -> str | None:
    data = await load_revenue_context(ctx)
    if not data:
        return None
    return format_renewal_context(data)

from framework.context.entity_resolvers import entity_resolver


@entity_resolver("crm")
async def resolve_crm_entities(refs: list[dict]) -> list[dict]:
    entities = []
    for ref in refs:
        entity = await crm_lookup(ref["type"], ref["id"])
        entities.append(
            {
                "system": "crm",
                "type": ref["type"],
                "id": ref["id"],
                "name": entity["name"],
                "content": format_entity_for_prompt(entity),
            }
        )
    return entities

from framework.knowledge.system_kbs import system_kb


system_kb(
    id="550e8400-e29b-41d4-a716-446655440000",
    name="Revenue Playbook",
    description="Internal revenue workflow guidance",
    agents=["RevenueAgent", "MainAgent"],
    source_path="knowledge/revenue_playbook.md",
    tags=["revenue", "playbook"],
    icon="book-open",
    icon_color="green",
)

uv run python scripts/inspect_agent.py MainAgent
uv run python scripts/inspect_artifact.py account_health
uv run pytest tests/unit/framework/test_prompt_discovery.py tests/unit/framework/test_artifact_discovery.py

GET /api/v1/agents/MainAgent/manifest