get_pipeline_status
Retrieve the current status of the book chapter pipeline to determine which guarded step (writer, editor, third-pass, or approve) should execute next.
Instructions
Next MCP tool to call per chapter (guarded pipeline hints).
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
No arguments | |||
Output Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| result | Yes |
Implementation Reference
- ChapterStatus enum defining the pipeline states (NOT_STARTED, DRAFT, AWAITING_EDITOR, AWAITING_THIRD_PASS, APPROVED) used by get_pipeline_status to determine next steps.
class ChapterStatus(str, Enum): NOT_STARTED = "not_started" DRAFT = "draft" AWAITING_EDITOR = "awaiting_editor" AWAITING_THIRD_PASS = "awaiting_third_pass" APPROVED = "approved" - src/storywright_mcp/workflow.py:110-138 (handler)Core handler logic for get_pipeline_status: iterates over all chapters and determines the next suggested MCP tool call based on each chapter's status. Delegates to ChapterStatus enum for state transitions.
def get_pipeline_status() -> str: """Human-readable next-step hints for every chapter + global checks.""" proj, _ = require_project() env = check_environment() lines = [env, "\n## Pipeline\n\n"] if not proj.chapters: lines.append("No chapters — use **add_chapter**.\n") return "".join(lines) for ch in sorted(proj.chapters, key=lambda c: c.num): st = ch.status if st == ChapterStatus.NOT_STARTED: nxt = f"`start_chapter({ch.num})`" elif st == ChapterStatus.DRAFT: nxt = f"`run_writer_agent({ch.num})`" elif st == ChapterStatus.AWAITING_EDITOR: nxt = f"`run_editor_review({ch.num})`" elif st == ChapterStatus.AWAITING_THIRD_PASS: missing = [a for a in proj.third_agents if a not in ch.third_pass_completed] if not proj.third_agents: nxt = f"`approve_chapter({ch.num})`" elif missing: nxt = f"`run_third_agent({ch.num}, agent_type=\"{missing[0]}\")` then others: {missing}" else: nxt = f"`approve_chapter({ch.num})`" else: nxt = "done" lines.append(f"- **Ch.{ch.num}** ({ch.status.value}) → {nxt}\n") return "".join(lines) - src/storywright_mcp/app.py:68-74 (registration)MCP tool registration via @mcp.tool() decorator. The async function delegates to workflow.get_pipeline_status() with a ValueError catch.
@mcp.tool() async def get_pipeline_status() -> str: """Next MCP tool to call per chapter (guarded pipeline hints).""" try: return workflow.get_pipeline_status() except ValueError as e: return str(e)