Back to Introducing the Vidbyte SDK
July 27th, 2026

Content Writer Agent

Writing

Pipelines

Workflows

Structured output

8 min read

The problem

You already know the audience, the goal, and the points to hit — what costs the afternoon is the first draft, and the one-shot draft always misses two of the points.

How it goes today

  • You write a long freeform prompt, get a draft, and reread the brief to see what it skipped.
  • You ask for a fix, and the rewrite drops something that was fine before.
  • Every writer on the team prompts differently, so drafts are not comparable week to week.

What you'll build

  • A writer agent that takes the same structured brief fields every time and returns typed markdown.
  • A reviewer agent that only checks the draft against the brief and reports what is missing.
  • A bounded loop that rewrites until the review passes, instead of you playing go-between.

Done looks like

  • The same brief fields go in every run, so two weeks of drafts are comparable.
  • The reviewer returns missing points as a list, not as an opinion paragraph.
  • A draft that fails review is rewritten automatically, up to a limit you set.
  • The run ends with a draft plus a checklist of what a human still has to verify.

Paste into Claude, Codex, Cursor, or another coding agent to scaffold this harness.

SDK example

Markdown
# Context: Vidbyte SDK Vidbyte SDK is a Python package (import as `vidbyte`) for building agent workflows yourself. Use it when you want to own the agent loop: system prompts, model/provider choice, tools, managed context, middleware, pipelines, tracing, evals, and multi-agent composition. Core mental model: 1. Create an Agent or BaseAgent. 2. Attach a system prompt, provider/model, optional tools, context, middleware, and runtime choices. 3. Call run() or arun(). 4. Let the SDK handle message assembly, tool schemas/calls, iteration limits, and optional pipelines. Public package boundary: reusable local agent infrastructure. It does not ship private Vidbyte learning models or proprietary platform internals.

Build it step by step

Every step adds to the same file. By the end you have one runnable writer.py.

  1. 01

    Decide what a draft is

    output_schema

    The checklist field matters as much as the markdown: it is where the model puts claims and links it could not verify, instead of quietly asserting them.

    + a typed draft with an editor checklist

    SDK example

    Python
    from pydantic import BaseModel, Field class Draft(BaseModel): title: str = Field(description="The headline for the piece.") markdown: str = Field(description="The full draft, ready for a human to edit.") editor_checklist: list[str] = Field(description="Claims or links a human must verify.")
  2. 02

    Pin down the inputs

    structured brief

    A structured brief is what makes two weeks of drafts comparable. Rendering it through one function means the writer sees the same fields every time, plus any notes from a previous attempt.

    + the same inputs every run

    SDK example

    Python
    brief = { "format": "blog post", "audience": "developers evaluating agent frameworks", "goal": "explain why a controllable agent loop matters", "tone": "clear, practical, non-hype", "length": "600-800 words", "must_include": ["tools and verification", "how it differs from one-shot chat", "one concrete example"], "must_avoid": ["fake customer logos", "unverified benchmarks"], } def format_brief(brief: dict, feedback: tuple[str, ...] = ()) -> str: """Render the brief, plus any reviewer notes from the previous attempt.""" lines = [f"{key}: {value}" for key, value in brief.items()] if feedback: lines.append("Fix these reviewer notes: " + "; ".join(feedback)) return "\n".join(lines)
  3. 03

    Build the writer

    Agent

    One agent with one job. Because it carries the Draft schema, its output is an object with a title and a body — not a blob you have to split apart.

    + the writer

    SDK example

    Python
    from vidbyte import Agent writer = Agent( name="content-writer", system_prompt=( "Write marketing and product drafts from a brief. " "Follow the audience, tone, format, and length exactly. " "Never invent product claims the brief did not give you." ), provider="openai", model_name="gpt-5.5", output_schema=Draft, )
  4. 04

    Build the reviewer

    second Agent + schema

    A separate agent that only compares the draft to the brief catches misses without rewriting everything. The approved boolean is what the loop in step 5 branches on.

    + the reviewer, and a machine-readable verdict

    SDK example

    Python
    class Review(BaseModel): missing_points: list[str] = Field(description="Must-include points the draft skipped.") tone_issues: list[str] = Field(description="Places the tone drifts from the brief.") invented_claims: list[str] = Field(description="Claims not supported by the brief.") approved: bool = Field(description="True only when nothing above needs fixing.") reviewer = Agent( name="brief-reviewer", system_prompt=( "Compare the draft to the brief. " "List missing points, tone mismatches, and invented claims. " "Approve only when all three lists are empty." ), provider="openai", model_name="gpt-5.5", output_schema=Review, )
  5. 05

    Chain them, if one pass is enough

    SequentialPipeline

    A pipeline threads one agent's text output into the next agent's prompt. It is the right shape when you just want a review printed alongside the draft — and if that is all you need, stop here.

    + the simple write-then-review chain

    SDK example

    Python
    from vidbyte import SequentialPipeline review_chain = SequentialPipeline([writer, reviewer]) printed_review = await review_chain.run(format_brief(brief))
  6. 06

    Loop until the review passes

    StateGraph

    A pipeline runs forward once and hands back text. When the review has to feed back into a rewrite, you need a graph: a stage that drafts, a validator that reads approved, and an edge that routes a rejection back to the stage.

    + a bounded rewrite loop

    SDK example

    Python
    from dataclasses import dataclass, field, replace from vidbyte import ( CallableStage, CallableValidator, MachineStatus, StageResult, StateGraph, StateMachineSettings, ValidationResult, ) @dataclass(frozen=True) class WriterState: brief: dict draft: Draft | None = None feedback: tuple[str, ...] = field(default_factory=tuple) async def draft_stage(ctx): # Writes a draft, then asks the reviewer to check it against the brief. prompt = format_brief(ctx.state.brief, feedback=ctx.state.feedback) written = (await writer.arun(prompt)).structured verdict = (await reviewer.arun(f"Brief: {ctx.state.brief}\n\nDraft:\n{written.markdown}")).structured notes = (*verdict.missing_points, *verdict.tone_issues, *verdict.invented_claims) return StageResult( replace(ctx.state, draft=written, feedback=notes), outcome="success" if verdict.approved else "needs_revision", ) def review_passed(ctx): # Passes only when the reviewer left no notes on the candidate draft. if not ctx.candidate_state.feedback: return ValidationResult.passed() return ValidationResult.rejected("needs_revision", "Address the reviewer notes.") graph = StateGraph(WriterState, name="draft-until-approved") graph.add_stage("draft", CallableStage(draft_stage), validators=(CallableValidator(review_passed),)) graph.add_terminal("approved", status=MachineStatus.SUCCEEDED) graph.set_entry("draft") graph.add_transition("draft", "approved") graph.add_transition("draft", "draft", on="needs_revision") machine = graph.compile(settings=StateMachineSettings(max_transitions=6))
  7. 07

    Run it on a real brief

    machine.arun

    The machine returns the last committed state along with how it terminated. Hitting the transition ceiling is a real outcome you should check, not an exception to ignore.

    + the run, and the approved draft

    SDK example

    Python
    result = await machine.arun(WriterState(brief=brief)) print("status:", result.status) print(result.state.draft.title) print(result.state.draft.markdown) print("verify before publishing:", result.state.draft.editor_checklist)
  8. 08

    The finished harness

    The seven steps above, in one file. The workflow is what runs; the pipeline from step 5 is kept as the one-pass alternative.

    writer.py — complete

    SDK example

    Python
    """Content Writer Agent - drafts from a brief and revises until a reviewer approves.""" import asyncio from dataclasses import dataclass, field, replace from pydantic import BaseModel, Field from vidbyte import ( Agent, CallableStage, CallableValidator, MachineStatus, StageResult, StateGraph, StateMachineSettings, ValidationResult, ) class Draft(BaseModel): title: str = Field(description="The headline for the piece.") markdown: str = Field(description="The full draft, ready for a human to edit.") editor_checklist: list[str] = Field(description="Claims or links a human must verify.") class Review(BaseModel): missing_points: list[str] = Field(description="Must-include points the draft skipped.") tone_issues: list[str] = Field(description="Places the tone drifts from the brief.") invented_claims: list[str] = Field(description="Claims not supported by the brief.") approved: bool = Field(description="True only when nothing above needs fixing.") writer = Agent( name="content-writer", system_prompt=( "Write marketing and product drafts from a brief. " "Follow the audience, tone, format, and length exactly. " "Never invent product claims the brief did not give you." ), provider="openai", model_name="gpt-5.5", output_schema=Draft, ) reviewer = Agent( name="brief-reviewer", system_prompt=( "Compare the draft to the brief. " "List missing points, tone mismatches, and invented claims. " "Approve only when all three lists are empty." ), provider="openai", model_name="gpt-5.5", output_schema=Review, ) brief = { "format": "blog post", "audience": "developers evaluating agent frameworks", "goal": "explain why a controllable agent loop matters", "tone": "clear, practical, non-hype", "length": "600-800 words", "must_include": ["tools and verification", "how it differs from one-shot chat", "one concrete example"], "must_avoid": ["fake customer logos", "unverified benchmarks"], } def format_brief(brief: dict, feedback: tuple[str, ...] = ()) -> str: """Render the brief, plus any reviewer notes from the previous attempt.""" lines = [f"{key}: {value}" for key, value in brief.items()] if feedback: lines.append("Fix these reviewer notes: " + "; ".join(feedback)) return "\n".join(lines) @dataclass(frozen=True) class WriterState: brief: dict draft: Draft | None = None feedback: tuple[str, ...] = field(default_factory=tuple) async def draft_stage(ctx): # Writes a draft, then asks the reviewer to check it against the brief. prompt = format_brief(ctx.state.brief, feedback=ctx.state.feedback) written = (await writer.arun(prompt)).structured verdict = (await reviewer.arun(f"Brief: {ctx.state.brief}\n\nDraft:\n{written.markdown}")).structured notes = (*verdict.missing_points, *verdict.tone_issues, *verdict.invented_claims) return StageResult( replace(ctx.state, draft=written, feedback=notes), outcome="success" if verdict.approved else "needs_revision", ) def review_passed(ctx): # Passes only when the reviewer left no notes on the candidate draft. if not ctx.candidate_state.feedback: return ValidationResult.passed() return ValidationResult.rejected("needs_revision", "Address the reviewer notes.") graph = StateGraph(WriterState, name="draft-until-approved") graph.add_stage("draft", CallableStage(draft_stage), validators=(CallableValidator(review_passed),)) graph.add_terminal("approved", status=MachineStatus.SUCCEEDED) graph.set_entry("draft") graph.add_transition("draft", "approved") graph.add_transition("draft", "draft", on="needs_revision") machine = graph.compile(settings=StateMachineSettings(max_transitions=6)) async def main() -> None: result = await machine.arun(WriterState(brief=brief)) print("status:", result.status) print(result.state.draft.title) print(result.state.draft.markdown) print("verify before publishing:", result.state.draft.editor_checklist) if __name__ == "__main__": asyncio.run(main())