sample_component_app.py

"""
E2E test: Custom session_table and memory_table via Components API.

1. Starts an AgentOS server with a workflow (no custom tables on the workflow itself)
2. Registers a workflow config via POST /components with custom table names
3. Runs the workflow via POST /workflows/{id}/runs
4. Checks Postgres to see if custom tables were created

This replicates the exact user flow from the issue.
"""

from agno.agent import Agent
from agno.db.postgres import PostgresDb
from agno.models.openai import OpenAIResponses
from agno.os import AgentOS
from agno.workflow import Step
from agno.workflow.workflow import Workflow

# ---------------------------------------------------------------------------
# DB
# ---------------------------------------------------------------------------
DB_URL = "postgresql+psycopg://ai:ai@localhost:5532/ai"
db = PostgresDb(db_url=DB_URL, db_schema="ai")

# ---------------------------------------------------------------------------
# A simple agent for the workflow step
# ---------------------------------------------------------------------------
news_agent = Agent(
    name="News Agent",
    model=OpenAIResponses(id="gpt-4.1-mini"),
    instructions="You are a helpful news agent. Summarize any topic briefly.",
    enable_user_memories=True,
    db=db
)

# ---------------------------------------------------------------------------
# Workflow (defined in code)
# ---------------------------------------------------------------------------
draft_workflow = Workflow(
    name="draft wf",
    description="draft wf",
    steps=[
        Step(
            name="News Agent",
            agent=news_agent,
        ),
    ],
    db=db,
)

# ---------------------------------------------------------------------------
# AgentOS
# ---------------------------------------------------------------------------
agent_os = AgentOS(
    name="Custom Table Test OS",
    workflows=[draft_workflow],
    agents=[news_agent],
    db=db,
)

app = agent_os.get_app()


if __name__ == "__main__":
    agent_os.serve(app="test:app", reload=True)