diff --git a/alembic/versions/2025_01_24_2003-26dc22efaf0b_add_status_to_workflow_table.py b/alembic/versions/2025_01_24_2003-26dc22efaf0b_add_status_to_workflow_table.py
new file mode 100644
index 00000000..9e09997f
--- /dev/null
+++ b/alembic/versions/2025_01_24_2003-26dc22efaf0b_add_status_to_workflow_table.py
@@ -0,0 +1,33 @@
+"""Add status to workflow table
+
+Revision ID: 26dc22efaf0b
+Revises: 3a37869686bd
+Create Date: 2025-01-24 20:03:14.509740+00:00
+
+"""
+
+from typing import Sequence, Union
+
+import sqlalchemy as sa
+
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision: str = "26dc22efaf0b"
+down_revision: Union[str, None] = "3a37869686bd"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.add_column("workflows", sa.Column("status", sa.String(), nullable=True))
+    op.execute("UPDATE workflows SET status = 'published'")
+    op.alter_column("workflows", "status", nullable=False)
+    # ### end Alembic commands ###
+
+
+def downgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_column("workflows", "status")
+    # ### end Alembic commands ###
diff --git a/skyvern/forge/sdk/db/client.py b/skyvern/forge/sdk/db/client.py
index c2024b37..6d40b966 100644
--- a/skyvern/forge/sdk/db/client.py
+++ b/skyvern/forge/sdk/db/client.py
@@ -81,6 +81,7 @@ from skyvern.forge.sdk.workflow.models.workflow import (
     WorkflowRunOutputParameter,
     WorkflowRunParameter,
     WorkflowRunStatus,
+    WorkflowStatus,
 )
 from skyvern.webeye.actions.actions import Action
 from skyvern.webeye.actions.models import AgentStepOutput
@@ -1090,6 +1091,7 @@ class AgentDB:
         workflow_permanent_id: str | None = None,
         version: int | None = None,
         is_saved_task: bool = False,
+        status: WorkflowStatus = WorkflowStatus.published,
     ) -> Workflow:
         async with self.Session() as session:
             workflow = WorkflowModel(
@@ -1103,6 +1105,7 @@ class AgentDB:
                 totp_identifier=totp_identifier,
                 persist_browser_session=persist_browser_session,
                 is_saved_task=is_saved_task,
+                status=status,
             )
             if workflow_permanent_id:
                 workflow.workflow_permanent_id = workflow_permanent_id
@@ -1177,6 +1180,7 @@ class AgentDB:
         only_saved_tasks: bool = False,
         only_workflows: bool = False,
         title: str = "",
+        statuses: list[WorkflowStatus] | None = None,
     ) -> list[Workflow]:
         """
         Get all workflows with the latest version for the organization.
@@ -1212,6 +1216,8 @@ class AgentDB:
                     main_query = main_query.where(WorkflowModel.is_saved_task.is_(False))
                 if title:
                     main_query = main_query.where(WorkflowModel.title.ilike(f"%{title}%"))
+                if statuses:
+                    main_query = main_query.where(WorkflowModel.status.in_(statuses))
                 main_query = (
                     main_query.order_by(WorkflowModel.created_at.desc()).limit(page_size).offset(db_page * page_size)
                 )
diff --git a/skyvern/forge/sdk/db/models.py b/skyvern/forge/sdk/db/models.py
index 9e9f456b..a2d1f08f 100644
--- a/skyvern/forge/sdk/db/models.py
+++ b/skyvern/forge/sdk/db/models.py
@@ -207,6 +207,7 @@ class WorkflowModel(Base):
     totp_verification_url = Column(String)
     totp_identifier = Column(String)
     persist_browser_session = Column(Boolean, default=False, nullable=False)
+    status = Column(String, nullable=False, default="published")
 
     created_at = Column(DateTime, default=datetime.datetime.utcnow, nullable=False)
     modified_at = Column(
diff --git a/skyvern/forge/sdk/db/utils.py b/skyvern/forge/sdk/db/utils.py
index 6e7a6d02..cd9f4b01 100644
--- a/skyvern/forge/sdk/db/utils.py
+++ b/skyvern/forge/sdk/db/utils.py
@@ -43,6 +43,7 @@ from skyvern.forge.sdk.workflow.models.workflow import (
     WorkflowRunOutputParameter,
     WorkflowRunParameter,
     WorkflowRunStatus,
+    WorkflowStatus,
 )
 
 LOG = structlog.get_logger()
@@ -187,6 +188,7 @@ def convert_to_workflow(workflow_model: WorkflowModel, debug_enabled: bool = Fal
         created_at=workflow_model.created_at,
         modified_at=workflow_model.modified_at,
         deleted_at=workflow_model.deleted_at,
+        status=WorkflowStatus(workflow_model.status),
     )
 
 
diff --git a/skyvern/forge/sdk/routes/agent_protocol.py b/skyvern/forge/sdk/routes/agent_protocol.py
index 1d80cd6a..2533e3db 100644
--- a/skyvern/forge/sdk/routes/agent_protocol.py
+++ b/skyvern/forge/sdk/routes/agent_protocol.py
@@ -69,6 +69,7 @@ from skyvern.forge.sdk.workflow.models.workflow import (
     WorkflowRun,
     WorkflowRunStatus,
     WorkflowRunStatusResponse,
+    WorkflowStatus,
 )
 from skyvern.forge.sdk.workflow.models.yaml import WorkflowCreateYAMLRequest
 from skyvern.webeye.actions.actions import Action
@@ -932,6 +933,7 @@ async def get_workflows(
         only_saved_tasks=only_saved_tasks,
         only_workflows=only_workflows,
         title=title,
+        statuses=[WorkflowStatus.published, WorkflowStatus.draft],
     )
 
 
@@ -1158,6 +1160,7 @@ async def observer_task(
             totp_verification_url=data.totp_verification_url,
             webhook_callback_url=data.webhook_callback_url,
             proxy_location=data.proxy_location,
+            publish_workflow=data.publish_workflow,
         )
     except LLMProviderError:
         LOG.error("LLM failure to initialize observer cruise", exc_info=True)
diff --git a/skyvern/forge/sdk/schemas/observers.py b/skyvern/forge/sdk/schemas/observers.py
index 758a5189..daa9c237 100644
--- a/skyvern/forge/sdk/schemas/observers.py
+++ b/skyvern/forge/sdk/schemas/observers.py
@@ -113,6 +113,7 @@ class ObserverTaskRequest(BaseModel):
     totp_verification_url: str | None = None
     totp_identifier: str | None = None
     proxy_location: ProxyLocation | None = None
+    publish_workflow: bool = False
 
     @field_validator("url", "webhook_callback_url", "totp_verification_url")
     @classmethod
diff --git a/skyvern/forge/sdk/services/observer_service.py b/skyvern/forge/sdk/services/observer_service.py
index 3ae14774..c29c9fb9 100644
--- a/skyvern/forge/sdk/services/observer_service.py
+++ b/skyvern/forge/sdk/services/observer_service.py
@@ -36,7 +36,13 @@ from skyvern.forge.sdk.workflow.models.block import (
     TaskBlock,
 )
 from skyvern.forge.sdk.workflow.models.parameter import PARAMETER_TYPE, ContextParameter
-from skyvern.forge.sdk.workflow.models.workflow import Workflow, WorkflowRequestBody, WorkflowRun, WorkflowRunStatus
+from skyvern.forge.sdk.workflow.models.workflow import (
+    Workflow,
+    WorkflowRequestBody,
+    WorkflowRun,
+    WorkflowRunStatus,
+    WorkflowStatus,
+)
 from skyvern.forge.sdk.workflow.models.yaml import (
     BLOCK_YAML_TYPES,
     PARAMETER_YAML_TYPES,
@@ -87,6 +93,7 @@ async def initialize_observer_cruise(
     totp_identifier: str | None = None,
     totp_verification_url: str | None = None,
     webhook_callback_url: str | None = None,
+    publish_workflow: bool = False,
 ) -> ObserverTask:
     observer_cruise = await app.DATABASE.create_observer_cruise(
         prompt=user_prompt,
@@ -127,8 +134,12 @@ async def initialize_observer_cruise(
     # create workflow and workflow run
     max_steps_override = 10
     try:
+        workflow_status = WorkflowStatus.published if publish_workflow else WorkflowStatus.auto_generated
         new_workflow = await app.WORKFLOW_SERVICE.create_empty_workflow(
-            organization, metadata.workflow_title, proxy_location=proxy_location
+            organization,
+            metadata.workflow_title,
+            proxy_location=proxy_location,
+            status=workflow_status,
         )
         workflow_run = await app.WORKFLOW_SERVICE.setup_workflow_run(
             request_id=None,
@@ -519,6 +530,7 @@ async def run_observer_cruise_helper(
             description=workflow.description,
             proxy_location=observer_cruise.proxy_location or ProxyLocation.RESIDENTIAL,
             workflow_definition=workflow_definition_yaml,
+            status=workflow.status,
         )
         LOG.info("Creating workflow from request", workflow_create_request=workflow_create_request)
         workflow = await app.WORKFLOW_SERVICE.create_workflow_from_request(
diff --git a/skyvern/forge/sdk/workflow/models/workflow.py b/skyvern/forge/sdk/workflow/models/workflow.py
index 0e347b61..d3f02e27 100644
--- a/skyvern/forge/sdk/workflow/models/workflow.py
+++ b/skyvern/forge/sdk/workflow/models/workflow.py
@@ -50,6 +50,12 @@ class WorkflowDefinition(BaseModel):
             raise WorkflowDefinitionHasDuplicateBlockLabels(duplicate_labels)
 
 
+class WorkflowStatus(StrEnum):
+    published = "published"
+    draft = "draft"
+    auto_generated = "auto_generated"
+
+
 class Workflow(BaseModel):
     workflow_id: str
     organization_id: str
@@ -64,6 +70,7 @@ class Workflow(BaseModel):
     totp_verification_url: str | None = None
     totp_identifier: str | None = None
     persist_browser_session: bool = False
+    status: WorkflowStatus = WorkflowStatus.published
 
     created_at: datetime
     modified_at: datetime
diff --git a/skyvern/forge/sdk/workflow/models/yaml.py b/skyvern/forge/sdk/workflow/models/yaml.py
index db984a9b..00fb87ba 100644
--- a/skyvern/forge/sdk/workflow/models/yaml.py
+++ b/skyvern/forge/sdk/workflow/models/yaml.py
@@ -7,6 +7,7 @@ from skyvern.config import settings
 from skyvern.forge.sdk.schemas.tasks import ProxyLocation
 from skyvern.forge.sdk.workflow.models.block import BlockType, FileType
 from skyvern.forge.sdk.workflow.models.parameter import ParameterType, WorkflowParameterType
+from skyvern.forge.sdk.workflow.models.workflow import WorkflowStatus
 
 
 class ParameterYAML(BaseModel, abc.ABC):
@@ -370,3 +371,4 @@ class WorkflowCreateYAMLRequest(BaseModel):
     persist_browser_session: bool = False
     workflow_definition: WorkflowDefinitionYAML
     is_saved_task: bool = False
+    status: WorkflowStatus = WorkflowStatus.published
diff --git a/skyvern/forge/sdk/workflow/service.py b/skyvern/forge/sdk/workflow/service.py
index 501c7465..e53ed337 100644
--- a/skyvern/forge/sdk/workflow/service.py
+++ b/skyvern/forge/sdk/workflow/service.py
@@ -73,6 +73,7 @@ from skyvern.forge.sdk.workflow.models.workflow import (
     WorkflowRunParameter,
     WorkflowRunStatus,
     WorkflowRunStatusResponse,
+    WorkflowStatus,
 )
 from skyvern.forge.sdk.workflow.models.yaml import (
     BLOCK_YAML_TYPES,
@@ -478,6 +479,7 @@ class WorkflowService:
         workflow_permanent_id: str | None = None,
         version: int | None = None,
         is_saved_task: bool = False,
+        status: WorkflowStatus = WorkflowStatus.published,
     ) -> Workflow:
         return await app.DATABASE.create_workflow(
             title=title,
@@ -492,6 +494,7 @@ class WorkflowService:
             workflow_permanent_id=workflow_permanent_id,
             version=version,
             is_saved_task=is_saved_task,
+            status=status,
         )
 
     async def get_workflow(self, workflow_id: str, organization_id: str | None = None) -> Workflow:
@@ -525,6 +528,7 @@ class WorkflowService:
         only_saved_tasks: bool = False,
         only_workflows: bool = False,
         title: str = "",
+        statuses: list[WorkflowStatus] | None = None,
     ) -> list[Workflow]:
         """
         Get all workflows with the latest version for the organization.
@@ -536,6 +540,7 @@ class WorkflowService:
             only_saved_tasks=only_saved_tasks,
             only_workflows=only_workflows,
             title=title,
+            statuses=statuses,
         )
 
     async def update_workflow(
@@ -1203,6 +1208,7 @@ class WorkflowService:
                     workflow_permanent_id=workflow_permanent_id,
                     version=existing_version + 1,
                     is_saved_task=request.is_saved_task,
+                    status=request.status,
                 )
             else:
                 workflow = await self.create_workflow(
@@ -1216,6 +1222,7 @@ class WorkflowService:
                     totp_identifier=request.totp_identifier,
                     persist_browser_session=request.persist_browser_session,
                     is_saved_task=request.is_saved_task,
+                    status=request.status,
                 )
             # Keeping track of the new workflow id to delete it if an error occurs during the creation process
             new_workflow_id = workflow.workflow_id
@@ -1707,7 +1714,11 @@ class WorkflowService:
         raise ValueError(f"Invalid block type {block_yaml.block_type}")
 
     async def create_empty_workflow(
-        self, organization: Organization, title: str, proxy_location: ProxyLocation | None = None
+        self,
+        organization: Organization,
+        title: str,
+        proxy_location: ProxyLocation | None = None,
+        status: WorkflowStatus = WorkflowStatus.published,
     ) -> Workflow:
         """
         Create a blank workflow with no blocks
@@ -1720,6 +1731,7 @@ class WorkflowService:
                 blocks=[],
             ),
             proxy_location=proxy_location,
+            status=status,
         )
         return await app.WORKFLOW_SERVICE.create_workflow_from_request(
             organization=organization,