refactor: remove schema helper API

README.md

@@ -4,7 +4,7 @@
 
 # Agent Client Protocol (Python)
 
-Build ACP-compliant agents and clients in Python with generated schema models, asyncio transports, helper builders, and runnable demos.
+Build ACP-compliant agents and clients in Python with generated schema models, asyncio transports, and runnable demos.
 
 > Releases track the upstream ACP schema; contributions that tighten coverage or tooling are always welcome.
 
@@ -19,14 +19,14 @@ uv add agent-client-protocol
 ## At a glance
 
 - **Spec parity:** Generated Pydantic models in `acp.schema` track every ACP release so payloads stay valid.
+- **Direct construction:** Generated discriminator fields have defaults, so models can be instantiated directly.
 - **Runtime ergonomics:** Async base classes, stdio JSON-RPC plumbing, and lifecycle helpers keep custom agents tiny.
 - **Examples ready:** Streaming, permissions, Gemini bridge, and duet demos live under `examples/`.
-- **Helper builders:** `acp.helpers` mirrors the Go/TS SDK APIs for content blocks, tool calls, and session updates.
 - **Contrib utilities:** Session accumulators, tool call trackers, and permission brokers share patterns from real deployments.
 
 ## Who benefits
 
-- Agent authors who need typed models, helper builders, and event-stream ergonomics for ACP-compatible assistants.
+- Agent authors who need typed models and event-stream ergonomics for ACP-compatible assistants.
 - Client integrators embedding ACP parties inside Python applications or wrapping existing CLIs via stdio.
 - Tooling teams experimenting with permission flows, streaming UX, or Gemini bridges without re-implementing transports.
 See real adopters like kimi-cli in the [Use Cases list](https://agentclientprotocol.github.io/python-sdk/use-cases/).
@@ -52,7 +52,7 @@ See real adopters like kimi-cli in the [Use Cases list](https://agentclientproto
 
 ## Project layout
 
-- `src/acp/`: runtime package (agents, clients, transports, helpers, schema bindings, contrib utilities)
+- `src/acp/`: runtime package (agents, clients, transports, schema bindings, contrib utilities)
 - `schema/`: upstream JSON schema sources (regenerate via `make gen-all`)
 - `docs/`: MkDocs content backing the published documentation
 - `examples/`: runnable scripts covering stdio orchestration patterns

docs/index.md

@@ -24,9 +24,8 @@ Next steps live in the [Quickstart](quickstart.md): launch the echo agent, wire
 
 ## SDK building blocks
 
-- `acp.schema`: generated Pydantic models that validate every payload against the canonical specification.
+- `acp.schema`: generated Pydantic models with defaults for protocol discriminator fields.
 - `acp.agent` / `acp.client`: async base classes, JSON-RPC supervision, and lifecycle orchestration.
-- `acp.helpers`: builders for content blocks, tool calls, permissions, and notifications.
 - `acp.contrib`: experimental utilities (session accumulators, permission brokers, tool call trackers) harvested from production deployments.
 - `examples/`: runnable agents, clients, duet demos, and the Gemini CLI bridge.
 
@@ -44,7 +43,7 @@ Next steps live in the [Quickstart](quickstart.md): launch the echo agent, wire
 ## Choose a path
 
 - **Just exploring?** Skim [use-cases.md](use-cases.md) to see how kimi-cli, agent-client-kernel, and others use the SDK.
-- **Building agents?** Copy `examples/echo_agent.py` or `examples/agent.py`, then layer in `acp.helpers` for tool calls and permissions.
+- **Building agents?** Copy `examples/echo_agent.py` or `examples/agent.py`, then use the generated models directly. Use `acp.contrib` for permission workflows and other stateful patterns.
 - **Embedding clients?** Start with `examples/client.py` or the `spawn_agent_process` / `spawn_client_process` helpers in the [Quickstart](quickstart.md#programmatic-launch).
 
 ## Reference material

docs/quickstart.md

@@ -93,8 +93,9 @@ import sys
 from pathlib import Path
 from typing import Any
 
-from acp import PROTOCOL_VERSION, spawn_agent_process, text_block
+from acp import PROTOCOL_VERSION, spawn_agent_process
 from acp.interfaces import Client
+from acp.schema import TextContentBlock
 
 
 class SimpleClient(Client):
@@ -114,7 +115,7 @@ async def main() -> None:
         session = await conn.new_session(cwd=str(script.parent), mcp_servers=[])
         await conn.prompt(
             session_id=session.session_id,
-            prompt=[text_block("Hello from spawn!")],
+            prompt=[TextContentBlock(text="Hello from spawn!")],
         )
 
 asyncio.run(main())
@@ -145,21 +146,17 @@ Run it with `run_agent()` inside an async entrypoint and wire it to your client.
 - [`examples/duet.py`](https://github.com/agentclientprotocol/python-sdk/blob/main/examples/duet.py) to see `spawn_agent_process` in action alongside the interactive client
 - [`examples/gemini.py`](https://github.com/agentclientprotocol/python-sdk/blob/main/examples/gemini.py) to drive the Gemini CLI (`--acp`; use `--experimental-acp` for older versions) directly from Python
 
-Need builders for common payloads? `acp.helpers` mirrors the Go/TS helper APIs:
+Generated models provide defaults for discriminator fields such as `type` and
+`sessionUpdate`, so they can be constructed directly:
 
 ```python
-from acp import start_tool_call, update_tool_call, text_block, tool_content
+from acp.schema import AgentMessageChunk, TextContentBlock
 
-start_update = start_tool_call("call-42", "Open file", kind="read", status="pending")
-finish_update = update_tool_call(
-    "call-42",
-    status="completed",
-    content=[tool_content(text_block("File opened."))],
+update = AgentMessageChunk(
+    content=TextContentBlock(text="File opened."),
 )
 ```
 
-Each helper wraps the generated Pydantic models in `acp.schema`, so the right discriminator fields (`type`, `sessionUpdate`, and friends) are always populated. That keeps examples readable while maintaining the same validation guarantees as constructing the models directly. Golden fixtures in `tests/test_golden.py` ensure the helpers stay in sync with future schema revisions.
-
 ## Optional — Talk to the Gemini CLI
 
 _Have the Gemini CLI installed? Run the bridge to exercise permission flows._

examples/agent.py

@@ -12,8 +12,6 @@
     PromptResponse,
     SetSessionModeResponse,
     run_agent,
-    text_block,
-    update_agent_message,
 )
 from acp.interfaces import Client
 from acp.schema import (
@@ -43,7 +41,7 @@ def on_connect(self, conn: Client) -> None:
         self._conn = conn
 
     async def _send_agent_message(self, session_id: str, content: Any) -> None:
-        update = content if isinstance(content, AgentMessageChunk) else update_agent_message(content)
+        update = content if isinstance(content, AgentMessageChunk) else AgentMessageChunk(content=content)
         await self._conn.session_update(session_id, update)
 
     async def initialize(
@@ -109,7 +107,7 @@ async def prompt(
         if session_id not in self._sessions:
             self._sessions.add(session_id)
 
-        await self._send_agent_message(session_id, text_block("Client sent:"))
+        await self._send_agent_message(session_id, TextContentBlock(text="Client sent:"))
         for block in prompt:
             await self._send_agent_message(session_id, block)
         return PromptResponse(stop_reason="end_turn")

examples/client.py

@@ -12,7 +12,6 @@
     Client,
     RequestError,
     connect_to_agent,
-    text_block,
 )
 from acp.core import ClientSideConnection
 from acp.schema import (
@@ -170,7 +169,7 @@ async def interactive_loop(conn: ClientSideConnection, session_id: str) -> None:
         try:
             await conn.prompt(
                 session_id=session_id,
-                prompt=[text_block(line)],
+                prompt=[TextContentBlock(text=line)],
             )
         except Exception as exc:
             logging.error("Prompt failed: %s", exc)  # noqa: TRY400

examples/echo_agent.py

@@ -14,11 +14,10 @@
     NewSessionResponse,
     PromptResponse,
     run_agent,
-    text_block,
-    update_agent_message,
 )
 from acp.interfaces import Client
 from acp.schema import (
+    AgentMessageChunk,
     AudioContentBlock,
     ClientCapabilities,
     EmbeddedResourceContentBlock,
@@ -70,7 +69,7 @@ async def prompt(
     ) -> PromptResponse:
         for block in prompt:
             text = block.get("text", "") if isinstance(block, dict) else getattr(block, "text", "")
-            chunk = update_agent_message(text_block(text))
+            chunk = AgentMessageChunk(content=TextContentBlock(text=text))
             chunk.field_meta = {"echo": True}
             chunk.content.field_meta = {"echo": True}
 

examples/gemini.py

@@ -17,7 +17,6 @@
     Client,
     RequestError,
     connect_to_agent,
-    text_block,
 )
 from acp.core import ClientSideConnection
 from acp.schema import (
@@ -254,7 +253,7 @@ def _print_text_content(content: object) -> None:
 async def _send_prompt(conn: ClientSideConnection, session_id: str, prompt: str, timeout: float | None) -> None:
     request = conn.prompt(
         session_id=session_id,
-        prompt=[text_block(prompt)],
+        prompt=[TextContentBlock(text=prompt)],
     )
     if timeout is None:
         await request

examples/http_client.py

@@ -12,9 +12,10 @@
 import asyncio
 from typing import Any
 
-from acp import connect_to_agent, text_block
+from acp import connect_to_agent
 from acp.http import create_http_stream
 from acp.interfaces import Client
+from acp.schema import TextContentBlock
 
 
 class ExampleClient(Client):
@@ -43,7 +44,7 @@ async def main() -> None:
         print(f"initialized (protocol v{init.protocol_version})")
         session = await conn.new_session(cwd=".", mcp_servers=[])
         print(f"session: {session.session_id}")
-        result = await conn.prompt(session_id=session.session_id, prompt=[text_block("hello over http")])
+        result = await conn.prompt(session_id=session.session_id, prompt=[TextContentBlock(text="hello over http")])
         print(f"stop reason: {result.stop_reason}")
     finally:
         await conn.close()

examples/http_server.py

@@ -25,12 +25,10 @@
     InitializeResponse,
     NewSessionResponse,
     PromptResponse,
-    text_block,
-    update_agent_message,
 )
 from acp.http.asgi import create_asgi_app
 from acp.interfaces import Client
-from acp.schema import ClientCapabilities, Implementation
+from acp.schema import AgentMessageChunk, ClientCapabilities, Implementation, TextContentBlock
 
 
 class EchoAgent(Agent):
@@ -56,7 +54,7 @@ async def prompt(self, session_id: str, prompt: list[Any], **kwargs: Any) -> Pro
             text = block.get("text", "") if isinstance(block, dict) else getattr(block, "text", "")
             await self._conn.session_update(
                 session_id=session_id,
-                update=update_agent_message(text_block(f"echo: {text}")),
+                update=AgentMessageChunk(content=TextContentBlock(text=f"echo: {text}")),
             )
         return PromptResponse(stop_reason="end_turn")
 

examples/ws_client.py

@@ -12,8 +12,9 @@
 import asyncio
 from typing import Any
 
-from acp import connect_to_agent, text_block
+from acp import connect_to_agent
 from acp.interfaces import Client
+from acp.schema import TextContentBlock
 from acp.ws import create_websocket_stream
 
 
@@ -42,7 +43,9 @@ async def main() -> None:
         print(f"initialized (protocol v{init.protocol_version})")
         session = await conn.new_session(cwd=".", mcp_servers=[])
         print(f"session: {session.session_id}")
-        result = await conn.prompt(session_id=session.session_id, prompt=[text_block("hello over websocket")])
+        result = await conn.prompt(
+            session_id=session.session_id, prompt=[TextContentBlock(text="hello over websocket")]
+        )
         print(f"stop reason: {result.stop_reason}")
     finally:
         await conn.close()

src/acp/__init__.py

@@ -7,31 +7,6 @@
     connect_to_agent,
     run_agent,
 )
-from .helpers import (
-    audio_block,
-    embedded_blob_resource,
-    embedded_text_resource,
-    image_block,
-    plan_entry,
-    resource_block,
-    resource_link_block,
-    session_notification,
-    start_edit_tool_call,
-    start_read_tool_call,
-    start_tool_call,
-    text_block,
-    tool_content,
-    tool_diff_content,
-    tool_terminal_ref,
-    update_agent_message,
-    update_agent_message_text,
-    update_agent_thought,
-    update_agent_thought_text,
-    update_plan,
-    update_tool_call,
-    update_user_message,
-    update_user_message_text,
-)
 from .meta import (
     AGENT_METHODS,
     CLIENT_METHODS,
@@ -198,30 +173,6 @@
     "spawn_client_process",
     "default_environment",
     "spawn_stdio_transport",
-    # helpers
-    "text_block",
-    "image_block",
-    "audio_block",
-    "resource_link_block",
-    "embedded_text_resource",
-    "embedded_blob_resource",
-    "resource_block",
-    "tool_content",
-    "tool_diff_content",
-    "tool_terminal_ref",
-    "plan_entry",
-    "update_plan",
-    "update_user_message",
-    "update_user_message_text",
-    "update_agent_message",
-    "update_agent_message_text",
-    "update_agent_thought",
-    "update_agent_thought_text",
-    "session_notification",
-    "start_tool_call",
-    "start_read_tool_call",
-    "start_edit_tool_call",
-    "update_tool_call",
 ]
 
 

src/acp/contrib/permissions.py

@@ -3,8 +3,14 @@
 from collections.abc import Awaitable, Callable, Sequence
 from typing import Any
 
-from ..helpers import text_block, tool_content
-from ..schema import PermissionOption, RequestPermissionRequest, RequestPermissionResponse, ToolCallUpdate
+from ..schema import (
+    ContentToolCallContent,
+    PermissionOption,
+    RequestPermissionRequest,
+    RequestPermissionResponse,
+    TextContentBlock,
+    ToolCallUpdate,
+)
 from .tool_calls import ToolCallTracker, _copy_model_list
 
 
@@ -75,7 +81,7 @@ async def request_for(
 
         if description:
             existing = tool_call.content or []
-            existing.append(tool_content(text_block(description)))
+            existing.append(ContentToolCallContent(content=TextContentBlock(text=description)))
             tool_call.content = existing
 
         option_set = tuple(option.model_copy(deep=True) for option in (options or self._default_options))

src/acp/contrib/tool_calls.py

@@ -6,8 +6,9 @@
 
 from pydantic import BaseModel, ConfigDict
 
-from ..helpers import text_block, tool_content
 from ..schema import (
+    ContentToolCallContent,
+    TextContentBlock,
     ToolCallLocation,
     ToolCallProgress,
     ToolCallStart,
@@ -173,7 +174,7 @@ def append_stream_text(
         status: Any = UNSET,
     ) -> ToolCallProgress:
         self._stream_buffer = (self._stream_buffer or "") + text
-        content = [tool_content(text_block(self._stream_buffer))]
+        content = [ContentToolCallContent(content=TextContentBlock(text=self._stream_buffer))]
         return self.update(title=title, status=status, content=content)