The control plane for agentic ETL and workflows.
Deploy, observe, and govern every agentic workflow your business runs on.

@ava.workflowdef brief_flow():    return (        load()        >> (extract() & research())        >> draft()        >> publish()    )

@ava.sourcedef load() -> ava.File:    return ava.File(load_file())

class ExtractItems(ava.Signature):    """Extract key facts and open questions.    Render every PDF page as an image and    delegate each page's visual review to subagents.    Inspect charts, tables, diagrams and small print;    do not rely on extracted text alone.    Merge findings with page references, deduplicate    facts and flag missing or conflicting evidence.    """    file: ava.File = ava.InputField()    key_items: KeyItems = ava.OutputField() @ava.agent_step(    ExtractItems, skills=[ava.agent.skills.pdf],)async def extract(    file: ava.File, *, agent: ava.Agent,) -> KeyItems:    result = await agent(file=file)    return result.key_items

class AnalyzeCompetition(ava.Signature):    """Use web_search to research relevant competitors.    Compare their positioning with the source document.    Cite source URLs, separate evidence from inference,    and flag gaps rather than inventing facts.    """    file: ava.File = ava.InputField()    analysis: CompetitiveAnalysis = ava.OutputField() @ava.agent_step(    AnalyzeCompetition,    skills=[ava.agent.skills.pdf],    tools=[web_search],)async def research(    file: ava.File, *, agent: ava.Agent,) -> CompetitiveAnalysis:    result = await agent(file=file)    return result.analysis

class DraftBrief(ava.Signature):    """Create a concise Word brief from both results.    Use the docx skill to write brief.docx with    key findings, competitive context, open questions    and citations. Return the generated Word file.    """    key_items: KeyItems = ava.InputField()    analysis: CompetitiveAnalysis = ava.InputField()    brief: ava.File = ava.OutputField() @ava.agent_step(    DraftBrief, skills=[ava.agent.skills.docx],)async def draft(    key_items: KeyItems,    analysis: CompetitiveAnalysis,    *, agent: ava.Agent,) -> ava.File:    result = await agent(        key_items=key_items, analysis=analysis,    )    return result.brief

@ava.destdef publish(brief: ava.File) -> None:    send_email(brief)