diff --git a/app/api/workflows.py b/app/api/workflows.py
new file mode 100644
index 0000000..a630bdb
--- /dev/null
+++ b/app/api/workflows.py
@@ -0,0 +1,336 @@
+"""工作流管理 API"""
+import uuid
+from datetime import datetime
+from typing import List, Optional
+
+from fastapi import APIRouter, HTTPException, BackgroundTasks
+from pydantic import BaseModel, Field
+
+from app.database import get_session
+from app.models import Workflow, WorkflowRun, WorkflowStep
+from app.workflow.models import Workflow as WorkflowModel, StepDef
+from app.workflow.engine import WorkflowEngine
+from app.config import get_config
+from app.llm.client import LLMClient
+from app.agent.engine import Agent as AgentEngine
+from app.agent.tools import global_registry
+from app.agent.memory import ConversationMemory
+
+router = APIRouter()
+
+
+class WorkflowCreate(BaseModel):
+    name: str
+    description: str = ""
+    steps: List[dict] = Field(default_factory=list)
+
+
+class WorkflowUpdate(BaseModel):
+    name: Optional[str] = None
+    description: Optional[str] = None
+    steps: Optional[List[dict]] = None
+    status: Optional[str] = None
+
+
+class WorkflowRunRequest(BaseModel):
+    context: dict = Field(default_factory=dict)
+
+
+def _build_agent_engine_for_workflow(agent_id: str) -> AgentEngine:
+    """从数据库加载 Agent 并构建引擎"""
+    session = get_session()
+    try:
+        from app.models import Agent as AgentModel
+        agent = session.query(AgentModel).filter(AgentModel.id == agent_id).first()
+        if not agent:
+            raise ValueError(f"Agent {agent_id} not found")
+
+        config = get_config()
+        provider_key = agent.llm_provider or "default"
+        provider_cfg = config.llm_providers.get(provider_key) or config.default_llm
+
+        llm = LLMClient(provider_cfg)
+        return AgentEngine(
+            name=agent.name,
+            llm_client=llm,
+            tools=global_registry,
+            memory=ConversationMemory(window_size=20),
+            system_prompt=agent.system_prompt or "You are a helpful assistant.",
+            temperature=agent.temperature,
+            max_tokens=agent.max_tokens,
+        )
+    finally:
+        session.close()
+
+
+@router.get("")
+async def list_workflows() -> List[dict]:
+    session = get_session()
+    try:
+        workflows = session.query(Workflow).order_by(Workflow.created_at.desc()).all()
+        return [
+            {
+                "id": w.id,
+                "name": w.name,
+                "description": w.description or "",
+                "status": w.status,
+                "step_count": len(w.definition.get("steps", [])) if w.definition else 0,
+                "created_at": w.created_at,
+                "updated_at": w.updated_at,
+            }
+            for w in workflows
+        ]
+    finally:
+        session.close()
+
+
+@router.post("")
+async def create_workflow(data: WorkflowCreate) -> dict:
+    session = get_session()
+    try:
+        wf = Workflow(
+            id=str(uuid.uuid4()),
+            name=data.name,
+            description=data.description,
+            definition={"steps": data.steps},
+            status="draft",
+        )
+        session.add(wf)
+        session.commit()
+        session.refresh(wf)
+        return {
+            "id": wf.id,
+            "name": wf.name,
+            "description": wf.description or "",
+            "status": wf.status,
+            "created_at": wf.created_at,
+        }
+    finally:
+        session.close()
+
+
+@router.get("/{workflow_id}")
+async def get_workflow(workflow_id: str) -> dict:
+    session = get_session()
+    try:
+        wf = session.query(Workflow).filter(Workflow.id == workflow_id).first()
+        if not wf:
+            raise HTTPException(404, "Workflow not found")
+        return {
+            "id": wf.id,
+            "name": wf.name,
+            "description": wf.description or "",
+            "definition": wf.definition or {},
+            "status": wf.status,
+            "created_at": wf.created_at,
+            "updated_at": wf.updated_at,
+        }
+    finally:
+        session.close()
+
+
+@router.put("/{workflow_id}")
+async def update_workflow(workflow_id: str, data: WorkflowUpdate) -> dict:
+    session = get_session()
+    try:
+        wf = session.query(Workflow).filter(Workflow.id == workflow_id).first()
+        if not wf:
+            raise HTTPException(404, "Workflow not found")
+
+        update_data = data.model_dump(exclude_unset=True)
+        if "steps" in update_data and update_data["steps"] is not None:
+            wf.definition = {"steps": update_data["steps"]}
+            update_data.pop("steps")
+
+        for key, value in update_data.items():
+            if value is not None:
+                setattr(wf, key, value)
+
+        session.commit()
+        session.refresh(wf)
+        return {
+            "id": wf.id,
+            "name": wf.name,
+            "description": wf.description or "",
+            "status": wf.status,
+            "updated_at": wf.updated_at,
+        }
+    finally:
+        session.close()
+
+
+@router.delete("/{workflow_id}")
+async def delete_workflow(workflow_id: str) -> dict:
+    session = get_session()
+    try:
+        wf = session.query(Workflow).filter(Workflow.id == workflow_id).first()
+        if not wf:
+            raise HTTPException(404, "Workflow not found")
+        session.delete(wf)
+        session.commit()
+        return {"deleted": workflow_id}
+    finally:
+        session.close()
+
+
+@router.post("/{workflow_id}/run")
+async def run_workflow(workflow_id: str, data: WorkflowRunRequest, background_tasks: BackgroundTasks) -> dict:
+    session = get_session()
+    try:
+        wf = session.query(Workflow).filter(Workflow.id == workflow_id).first()
+        if not wf:
+            raise HTTPException(404, "Workflow not found")
+
+        run = WorkflowRun(
+            id=str(uuid.uuid4()),
+            workflow_id=workflow_id,
+            status="running",
+        )
+        session.add(run)
+        session.commit()
+        session.refresh(run)
+
+        # Build workflow model
+        wf_model = Workflow(
+            name=wf.name,
+            description=wf.description or "",
+            steps=[],
+        )
+        for step_data in (wf.definition or {}).get("steps", []):
+            wf_model.steps.append(StepDef(**step_data))
+
+        # Start execution in background
+        background_tasks.add_task(
+            _execute_workflow,
+            run.id,
+            wf_model,
+            data.context,
+        )
+
+        return {"run_id": run.id, "status": "running"}
+    finally:
+        session.close()
+
+
+async def _execute_workflow(run_id: str, workflow: WorkflowModel, context: dict) -> None:
+    """后台执行工作流"""
+    session = get_session()
+    try:
+        run = session.query(WorkflowRun).filter(WorkflowRun.id == run_id).first()
+        if not run:
+            return
+        run.status = "running"
+        session.commit()
+
+        # Build agent engines
+        agent_engines = {}
+        for step in workflow.steps:
+            if step.agent_id and step.agent_id not in agent_engines:
+                try:
+                    agent_engines[step.agent_id] = _build_agent_engine_for_workflow(step.agent_id)
+                except Exception as e:
+                    run.status = "failed"
+                    run.error = f"Failed to load agent {step.agent_id}: {str(e)}"
+                    run.completed_at = datetime.utcnow()
+                    session.commit()
+                    return
+
+        # Simple step executor
+        def step_executor(step_name: str, agent_id: str, prompt: str, step_context: dict) -> str:
+            agent = agent_engines.get(agent_id)
+            if not agent:
+                return f"Error: Agent {agent_id} not available"
+            messages = [{"role": "user", "content": prompt}]
+            return agent.run(messages)
+
+        engine = WorkflowEngine()
+        result = engine.run(workflow, context, step_executor=step_executor)
+
+        # Save step results
+        for step_result in result.step_results:
+            step_record = WorkflowStep(
+                run_id=run_id,
+                step_name=step_result.step_name,
+                status=step_result.status.value,
+                input_data={"context_keys": list(context.keys())},
+                output_data={"output": step_result.output, "error": step_result.error},
+                completed_at=datetime.utcnow(),
+            )
+            session.add(step_record)
+
+        run.status = "completed" if result.all_completed else "failed"
+        run.results = result.to_dict()
+        if result.any_failed:
+            run.error = "One or more steps failed"
+        run.completed_at = datetime.utcnow()
+        session.commit()
+
+    except Exception as e:
+        session.query(WorkflowRun).filter(WorkflowRun.id == run_id).update({
+            "status": "failed",
+            "error": str(e),
+            "completed_at": datetime.utcnow(),
+        })
+        session.commit()
+    finally:
+        session.close()
+
+
+@router.get("/{workflow_id}/runs")
+async def list_workflow_runs(workflow_id: str) -> List[dict]:
+    session = get_session()
+    try:
+        runs = (
+            session.query(WorkflowRun)
+            .filter(WorkflowRun.workflow_id == workflow_id)
+            .order_by(WorkflowRun.started_at.desc())
+            .all()
+        )
+        return [
+            {
+                "id": r.id,
+                "workflow_id": r.workflow_id,
+                "status": r.status,
+                "started_at": r.started_at,
+                "completed_at": r.completed_at,
+                "error": r.error,
+            }
+            for r in runs
+        ]
+    finally:
+        session.close()
+
+
+@router.get("/runs/{run_id}")
+async def get_workflow_run(run_id: str) -> dict:
+    session = get_session()
+    try:
+        run = session.query(WorkflowRun).filter(WorkflowRun.id == run_id).first()
+        if not run:
+            raise HTTPException(404, "Workflow run not found")
+
+        steps = [
+            {
+                "id": s.id,
+                "step_name": s.step_name,
+                "status": s.status,
+                "input_data": s.input_data,
+                "output_data": s.output_data,
+                "started_at": s.started_at,
+                "completed_at": s.completed_at,
+            }
+            for s in run.steps
+        ]
+
+        return {
+            "id": run.id,
+            "workflow_id": run.workflow_id,
+            "status": run.status,
+            "results": run.results,
+            "error": run.error,
+            "started_at": run.started_at,
+            "completed_at": run.completed_at,
+            "steps": steps,
+        }
+    finally:
+        session.close()