Agent Client Protocol (ACP) implementation for Python

These details have not been verified by PyPI

Project links

Project description

chuk-acp

A Python implementation of the Agent Client Protocol (ACP) - the standard protocol for communication between code editors and AI coding agents.

📖 Table of Contents

Overview
Why ACP?
Features
Installation
Quick Start
Core Concepts
Complete Examples
API Reference
Protocol Support
Architecture
Testing
Relationship to MCP
Contributing
License
Links

Overview

The Agent Client Protocol (ACP) is to AI coding agents what the Language Server Protocol (LSP) is to programming languages. It standardizes communication between code editors/IDEs and coding agents—programs that use generative AI to autonomously modify code.

chuk-acp provides a complete, production-ready Python implementation of ACP, enabling you to:

💬 Interact with agents instantly using the CLI (uvx chuk-acp claude-code-acp or echo "hello" | uvx chuk-acp client -- kimi --acp)
🤖 Build ACP-compliant coding agents easily with the high-level ACPAgent API
🖥️ Build editors/IDEs that can connect to any ACP-compliant agent with ACPClient
🔌 Integrate AI capabilities into existing development tools
🧪 Test and develop against the ACP specification

Why ACP?

The Problem

Without a standard protocol, every AI coding tool creates its own proprietary interface, leading to:

Fragmentation across different tools and editors
Inability to switch agents or editors without rewriting integration code
Duplicated effort implementing similar functionality
Limited interoperability

The Solution

ACP provides a standard, open protocol that:

✅ Enables any agent to work with any editor
✅ Provides consistent user experience across tools
✅ Allows innovation at both the editor and agent level
✅ Built on proven standards (JSON-RPC 2.0)
✅ Supports async/streaming for real-time AI interactions

Think LSP for language tooling, but for AI coding agents.

Features

🎯 Complete ACP Implementation

Full support for ACP v1 specification
All baseline methods and content types
Optional capabilities (modes, session loading, file system, terminal)
Protocol compliance test suite
MCP Servers Support: Automatically sends empty mcpServers: [] for compatibility with agents that require it
Flexible AgentInfo: Handles agents that return incomplete initialization data

🔧 Developer-Friendly

CLI Tool: Interactive command-line client for testing agents (uvx chuk-acp)
Zero Installation: Run with uvx - no setup required
Type-Safe: Comprehensive type hints throughout
Async-First: Built on anyio for efficient async/await patterns
Optional Pydantic: Use Pydantic for validation, or go dependency-free with fallback
Well-Documented: Extensive examples and API documentation
Production-Ready: Tested across Python 3.11, 3.12 on Linux, macOS, Windows

🚀 Flexible & Extensible

Multiple transports: Stdio (with more coming)
Custom methods: Extend protocol with _meta fields and custom methods
Pluggable: Easy to integrate into existing tools
MCP Integration: Seamless compatibility with Model Context Protocol

🛡️ Quality & Security

Comprehensive test coverage
Security scanning with Bandit and CodeQL
Type checking with mypy
Automated dependency updates
CI/CD with GitHub Actions

Quick Start (60 Seconds!)

🚀 Try Without Installation

The absolute fastest way to get started - no cloning, no installation:

# 1. Download a standalone agent example
curl -O https://raw.githubusercontent.com/chuk-ai/chuk-acp/main/examples/standalone_agent.py

# 2. Run it with uvx (automatically installs chuk-acp temporarily)
uvx chuk-acp client python standalone_agent.py

# That's it! Start chatting with your agent.

See QUICKSTART.md for full details.

Or Connect to External Agents

# Claude Code (requires ANTHROPIC_API_KEY)
ANTHROPIC_API_KEY=sk-... uvx chuk-acp claude-code-acp

# Kimi (Chinese AI assistant)
uvx chuk-acp client -- kimi --acp

That's it! uvx automatically handles installation. Perfect for quick testing.

Installation

Using uvx (Recommended for One-Off Usage)

No installation needed! uvx runs the CLI directly:

# Single prompt mode (interactive with stdin)
echo "Create a Python function to calculate fibonacci" | uvx chuk-acp client -- kimi --acp

# With faster validation (optional)
uvx --from 'chuk-acp[pydantic]' chuk-acp claude-code-acp

Using uv (Recommended for Development)

uv is a fast Python package installer:

# Basic installation (includes CLI)
uv pip install chuk-acp

# With Pydantic validation support (recommended for better performance)
uv pip install chuk-acp[pydantic]

# Or add to your project
uv add chuk-acp

Using pip

# Basic installation (includes CLI)
pip install chuk-acp

# With Pydantic support (recommended)
pip install chuk-acp[pydantic]

Development Installation

git clone https://github.com/chuk-ai/chuk-acp.git
cd chuk-acp
uv venv
source .venv/bin/activate  # On Windows: .venv\Scripts\activate
uv pip install -e ".[dev,pydantic]"

Requirements

Python 3.11 or higher
Dependencies: anyio, typing-extensions
Optional: pydantic (for faster validation - works without it using fallback mechanism)

Quick Start

CLI Tool - Interactive Chat with Any Agent

The easiest way to interact with ACP agents is using the built-in CLI. Works instantly with uvx or after installation.

The CLI supports three modes:

Agent passthrough - For editors/tools (explicitly use agent subcommand)
Interactive client - For humans (explicitly use client subcommand)
Auto-detect - Smart mode selection (TTY = interactive, piped = passthrough)

Try It Now (No Installation!)

# Connect to Claude Code (requires ANTHROPIC_API_KEY)
ANTHROPIC_API_KEY=sk-... uvx chuk-acp client claude-code-acp

# Connect to Kimi agent (use -- to separate flags)
echo "hello" | uvx chuk-acp client -- kimi --acp

# Interactive mode with verbose output
echo "your question" | uvx chuk-acp --verbose client -- kimi --acp

# Interactive chat opens automatically
# Just start typing your questions!

After Installation

# Interactive client mode (explicit)
chuk-acp client python examples/echo_agent.py

# Agent passthrough mode (for editors like Zed)
chuk-acp agent python my_agent.py

# Auto-detect mode (interactive if TTY, passthrough if piped)
chuk-acp python examples/echo_agent.py

# Single prompt (requires stdin for kimi)
echo "Create a Python function to calculate factorial" | chuk-acp client -- kimi --acp

# Using a config file
chuk-acp --config examples/kimi_config.json

# With environment variables
chuk-acp client python agent.py --env DEBUG=true --env API_KEY=xyz

# Force interactive mode even when piped
chuk-acp --interactive python agent.py

# Verbose output for debugging
chuk-acp client python agent.py --verbose

Interactive Mode Commands

When in interactive chat mode, you can use these special commands:

Command	Description
`/quit` or `/exit`	Exit the client
`/new`	Start a new session (clears context)
`/info`	Show agent information and session ID

Example Interactive Session

With Claude Code:

$ ANTHROPIC_API_KEY=sk-... uvx chuk-acp claude-code-acp

╔═══════════════════════════════════════════════════════════════╗
║           ACP Interactive Client                              ║
╚═══════════════════════════════════════════════════════════════╝

You: Create a Python function to check if a string is a palindrome

Agent: Here's a Python function to check if a string is a palindrome:

def is_palindrome(s):
    # Remove spaces and convert to lowercase
    s = s.replace(" ", "").lower()
    # Check if string equals its reverse
    return s == s[::-1]

You: /quit
Goodbye!

With Kimi:

$ echo "What's the best way to handle async errors in Python?" | uvx chuk-acp client -- kimi --acp

Agent: [Kimi's response with thinking and detailed explanation...]

Configuration Files

Use standard ACP configuration format (compatible with Zed, VSCode, etc.):

claude_code_config.json:

{
  "command": "claude-code-acp",
  "args": [],
  "env": {
    "ANTHROPIC_API_KEY": "sk-..."
  }
}

kimi_config.json:

{
  "command": "kimi",
  "args": ["--acp"],
  "env": {}
}

Then use with:

chuk-acp --config claude_code_config.json
chuk-acp --config kimi_config.json

Important: Using `--` for Agent Flags

When an agent requires flags (like kimi --acp), use -- to separate chuk-acp flags from agent arguments:

# Correct - use -- separator
chuk-acp client -- kimi --acp

# Also works via config file
chuk-acp --config kimi_config.json

Without --, argparse treats --acp as a chuk-acp flag and fails. Using a config file avoids this issue entirely.

📖 See CLI.md for complete CLI documentation and advanced usage.

Using with Editors (Zed, VSCode, etc.)

chuk-acp can run your custom agents in editors that support ACP, like Zed. The agent mode provides a passthrough that lets editors communicate directly with your agent via the ACP protocol.

Zed Configuration

Add this to your Zed settings.json (usually ~/.config/zed/settings.json):

{
  "agent_servers": {
    "My Custom Agent": {
      "command": "uvx",
      "args": [
        "chuk-acp",
        "agent",
        "python",
        "/absolute/path/to/my_agent.py"
      ],
      "env": {}
    }
  }
}

Or using auto-detect mode (will automatically use passthrough when piped from Zed):

{
  "agent_servers": {
    "My Custom Agent": {
      "command": "uvx",
      "args": [
        "chuk-acp",
        "python",
        "/absolute/path/to/my_agent.py"
      ],
      "env": {}
    }
  }
}

What this does:

uvx automatically installs chuk-acp in a temporary environment
chuk-acp agent runs in passthrough mode (just executes your agent with the ACP protocol layer)
Your agent gets access to the chuk-acp package (so from chuk_acp.agent import ACPAgent works)
No --from or --with flags needed!

Example agent file (my_agent.py):

from typing import List
from chuk_acp.agent import ACPAgent, AgentSession
from chuk_acp.protocol.types import AgentInfo, Content

class MyAgent(ACPAgent):
    def get_agent_info(self) -> AgentInfo:
        return AgentInfo(name="my-agent", version="1.0.0")

    async def handle_prompt(self, session: AgentSession, prompt: List[Content]) -> str:
        text = prompt[0].get("text", "") if prompt else ""
        return f"You said: {text}"

if __name__ == "__main__":
    MyAgent().run()

See examples/zed_config.json for more configuration examples.

The Easiest Way: ACPClient

The fastest way to get started programmatically is with the high-level ACPClient, which handles all protocol details automatically:

Option A: Direct Usage

"""quickstart.py"""
import anyio
from chuk_acp import ACPClient

async def main():
    # Connect to an agent - handles initialization, sessions, everything!
    async with ACPClient("python", ["echo_agent.py"]) as client:
        # Send a prompt and get the response
        result = await client.send_prompt("Hello!")
        print(f"Agent: {result.full_message}")

anyio.run(main)

Option B: Using Standard ACP Configuration

This matches the configuration format used by editors like Zed, VSCode, etc.:

"""quickstart_config.py"""
import anyio
from chuk_acp import ACPClient, AgentConfig

async def main():
    # Standard ACP configuration format
    config = AgentConfig(
        command="kimi",           # Any ACP-compatible agent
        args=["--acp"],          # Agent-specific arguments
        env={"DEBUG": "true"}    # Optional environment variables
    )

    async with ACPClient.from_config(config) as client:
        result = await client.send_prompt("Hello!")
        print(f"Agent: {result.full_message}")

anyio.run(main)

Or load from a JSON file (like ~/.config/zed/settings.json):

from chuk_acp import load_agent_config

config = load_agent_config("~/.config/my-app/agent.json")
async with ACPClient.from_config(config) as client:
    result = await client.send_prompt("Hello!")

What ACPClient does automatically:

✅ Starts the agent process
✅ Handles protocol initialization
✅ Creates and manages sessions
✅ Captures all notifications
✅ Cleans up resources
✅ Supports standard ACP configuration format

Want more control? The low-level API gives you fine-grained control over the protocol. See the examples below.

Building an Agent

The fastest way to build an ACP agent is with the high-level ACPAgent class:

"""my_agent.py"""
from typing import List
from chuk_acp.agent import ACPAgent, AgentSession
from chuk_acp.protocol.types import AgentInfo, Content

class MyAgent(ACPAgent):
    """Your custom agent implementation."""

    def get_agent_info(self) -> AgentInfo:
        """Return agent information."""
        return AgentInfo(
            name="my-agent",
            version="1.0.0",
            title="My Custom Agent"
        )

    async def handle_prompt(
        self, session: AgentSession, prompt: List[Content]
    ) -> str:
        """Handle a prompt - this is where your agent logic goes."""
        # Extract text from prompt
        text = prompt[0].get("text", "") if prompt else ""

        # Your agent logic here
        response = f"I received: {text}"

        # Return the response
        return response

if __name__ == "__main__":
    agent = MyAgent()
    agent.run()

Run your agent:

# Test with CLI (using uv for environment management)
uvx chuk-acp client uv run my_agent.py

# Or if chuk-acp is installed globally
chuk-acp client python my_agent.py

What ACPAgent does automatically:

✅ Handles all protocol messages (initialize, session/new, session/prompt)
✅ Manages sessions and routing
✅ Sends responses in correct format
✅ Error handling and logging
✅ Stdin/stdout transport

Real example: See examples/echo_agent.py - a complete working agent in just 35 lines!

🚀 Even Easier: chuk-acp-agent

For an even simpler agent-building experience, check out chuk-acp-agent - an opinionated agent kit built on top of chuk-acp:

from chuk_acp_agent import Agent, Context

class MyAgent(Agent):
    async def on_prompt(self, ctx: Context, prompt: str):
        # Session memory
        count = ctx.memory.get("count", 0) + 1
        ctx.memory.set("count", count)

        # Stream response
        yield f"Message #{count}: {prompt}\n"

if __name__ == "__main__":
    MyAgent().run()

Features:

📦 Batteries-included: session memory, streaming helpers, MCP integration
🎯 Minimal boilerplate: just implement on_prompt()
🔧 Context API: ctx.memory, ctx.emit(), ctx.fs, ctx.terminal
🛠️ Built-in MCP tool support via chuk-tool-processor

Install: pip install chuk-acp-agent or see the chuk-acp-agent docs.

More Examples

For more complete examples showing different use cases:

# Clone the repository
git clone https://github.com/chuk-ai/chuk-acp.git
cd chuk-acp

# Install
uv pip install -e ".[pydantic]"

# Run examples (all use the high-level ACPClient)
uv run python examples/simple_client.py   # Basic single prompt
uv run python examples/quick_start.py     # Multi-turn conversation
uv run python examples/config_example.py  # Configuration support (Zed/VSCode format)

# Advanced: Low-level protocol examples
uv run python examples/low_level/simple_client.py        # Manual protocol handling
uv run python examples/low_level/quick_start.py          # Self-contained with embedded agent
uv run python examples/low_level/comprehensive_demo.py   # All ACP features

See the examples directory for detailed documentation.

Note: Examples are in the GitHub repository. If you installed via pip, clone the repo to access them.

Option B: Build Your Own (10 Minutes)

Create a complete ACP client and agent from scratch.

Step 1: Install

uv pip install chuk-acp[pydantic]

Step 2: Create an Agent

Save this as echo_agent.py:

"""echo_agent.py - A simple ACP agent"""
import json
import sys
import uuid

from chuk_acp.protocol import (
    create_response,
    create_notification,
    METHOD_INITIALIZE,
    METHOD_SESSION_NEW,
    METHOD_SESSION_PROMPT,
    METHOD_SESSION_UPDATE,
)
from chuk_acp.protocol.types import AgentInfo, AgentCapabilities, TextContent

# Read messages from stdin, write to stdout
for line in sys.stdin:
    msg = json.loads(line.strip())
    method = msg.get("method")
    params = msg.get("params", {})
    msg_id = msg.get("id")

    # Route to handlers
    if method == METHOD_INITIALIZE:
        result = {
            "protocolVersion": 1,
            "agentInfo": AgentInfo(name="echo-agent", version="1.0.0").model_dump(),
            "agentCapabilities": AgentCapabilities().model_dump(),
        }
        response = create_response(id=msg_id, result=result)

    elif method == METHOD_SESSION_NEW:
        session_id = f"session-{uuid.uuid4().hex[:8]}"
        response = create_response(id=msg_id, result={"sessionId": session_id})

    elif method == METHOD_SESSION_PROMPT:
        session_id = params["sessionId"]
        user_text = params["prompt"][0].get("text", "")

        # Send a notification with the echo
        from chuk_acp.protocol.types import SessionUpdate

        session_update = SessionUpdate(
            sessionUpdate="agent_message_chunk",
            content=TextContent(text=f"Echo: {user_text}")
        )
        notification = create_notification(
            method=METHOD_SESSION_UPDATE,
            params={
                "sessionId": session_id,
                "update": session_update.model_dump(exclude_none=True),
            },
        )
        sys.stdout.write(json.dumps(notification.model_dump()) + "\n")
        sys.stdout.flush()

        # Send the response
        response = create_response(id=msg_id, result={"stopReason": "end_turn"})

    else:
        continue

    sys.stdout.write(json.dumps(response.model_dump()) + "\n")
    sys.stdout.flush()

Step 3: Create a Client

Save this as my_client.py:

"""my_client.py - Connect to the echo agent using ACPClient"""
import anyio
from chuk_acp import ACPClient


async def main():
    # Connect to the agent - handles everything automatically!
    async with ACPClient("python", ["echo_agent.py"]) as client:
        # Send a prompt and get the response
        result = await client.send_prompt("Hello!")
        print(f"Agent says: {result.full_message}")


if __name__ == "__main__":
    anyio.run(main())

Step 4: Run It!

uv run python my_client.py

Output:

✓ Connected to echo-agent
✓ Session: session-a1b2c3d4

Sending: Hello!
Agent says: Echo: Hello!
✓ Done!

🎉 That's it! You've built a working ACP agent and client.

What You Learned

Option A showed you the fastest path - running pre-built examples.

Option B taught you:

Agents: Read JSON-RPC from stdin, write to stdout using create_response() and create_notification()
Clients: Connect via stdio_transport, use send_initialize() and send_session_new(), manually handle messages to capture notifications
Protocol flow: Initialize → Create Session → Send Prompts (with notifications) → Get Response
Best practices: Use library types (TextContent, AgentInfo) and method constants (METHOD_INITIALIZE)

Next Steps

Explore More Features:

Check out the complete examples in the GitHub repository:

simple_client.py - Clean client with notification handling
echo_agent.py - Production-ready agent with error handling
comprehensive_demo.py - Filesystem, terminal, all ACP features

Build Something:

Add file system access to your agent (see Example 3 below)
Implement tool calls and permission requests
Support multiple concurrent sessions
Add streaming for long responses

Learn More:

API Reference - Complete API documentation
Protocol Support - What's supported in ACP v1
ACP Specification - Official protocol docs

Core Concepts

The Agent-Client Model

┌─────────────┐                  ┌──────────────┐
│   Client    │ ←── JSON-RPC ──→ │    Agent     │
│  (Editor)   │     over stdio   │  (AI Tool)   │
└─────────────┘                  └──────────────┘
      ↑                                 ↑
      │                                 │
  User Interface                   AI Model
  File System                      Code Analysis
  Permissions                      Code Generation

Key Components

1. Protocol Layer (`chuk_acp.protocol`)

The core protocol implementation:

JSON-RPC 2.0: Request/response and notification messages
Message Types: Initialize, session management, prompts
Content Types: Text, images, audio, resources, annotations
Capabilities: Negotiate features between client and agent

2. Transport Layer (`chuk_acp.transport`)

Communication mechanism:

Stdio Transport: Process-based communication (current)
Extensible: WebSocket, HTTP, etc. (future)

3. Type System (`chuk_acp.protocol.types`)

Strongly-typed protocol structures:

Content types (text, image, audio)
Capabilities and features
Session modes and states
Tool calls and permissions

The ACP Flow

1. INITIALIZE
   Client ──→ Agent: Protocol version, capabilities
   Agent  ──→ Client: Agent info, supported features

2. SESSION CREATION
   Client ──→ Agent: Working directory, MCP servers
   Agent  ──→ Client: Session ID

3. PROMPT TURN
   Client ──→ Agent: User prompt (text, images, etc.)
   Agent  ──→ Client: [Streaming updates]
   Agent  ──→ Client: Stop reason (end_turn, max_tokens, etc.)

4. ONGOING INTERACTION
   - Session updates (thoughts, tool calls, messages)
   - Permission requests (file access, terminal, etc.)
   - Mode changes (ask → code → architect)
   - Cancellation support

Complete Examples

Example 1: Echo Agent (Using Library)

A minimal agent that echoes user input using chuk-acp library helpers:

"""echo_agent.py - Agent using chuk-acp library"""
import json
import sys
import uuid
from typing import Dict, Any

from chuk_acp.protocol import (
    create_response,
    create_error_response,
    create_notification,
    METHOD_INITIALIZE,
    METHOD_SESSION_NEW,
    METHOD_SESSION_PROMPT,
    METHOD_SESSION_UPDATE,
)
from chuk_acp.protocol.types import (
    AgentInfo,
    AgentCapabilities,
    SessionUpdate,
    TextContent,
)

class EchoAgent:
    def __init__(self):
        self.sessions = {}

    def handle_initialize(self, params: Dict[str, Any]) -> Dict[str, Any]:
        """Use library types instead of manual dict construction."""
        agent_info = AgentInfo(name="echo-agent", version="0.1.0")
        agent_capabilities = AgentCapabilities()

        return {
            "protocolVersion": 1,
            "agentInfo": agent_info.model_dump(exclude_none=True),
            "agentCapabilities": agent_capabilities.model_dump(exclude_none=True),
        }

    def handle_session_new(self, params: Dict[str, Any]) -> Dict[str, Any]:
        session_id = f"session_{uuid.uuid4().hex[:8]}"
        self.sessions[session_id] = {"cwd": params.get("cwd")}
        return {"sessionId": session_id}

    def handle_session_prompt(self, params: Dict[str, Any]) -> Dict[str, Any]:
        session_id = params["sessionId"]
        prompt = params["prompt"]

        # Use library helpers to create notification
        text_content = TextContent(
            text=f"Echo: You said '{prompt[0].get('text', '')}'"
        )

        session_update = SessionUpdate(
            sessionUpdate="agent_message_chunk",
            content=text_content
        )

        notification = create_notification(
            method=METHOD_SESSION_UPDATE,
            params={
                "sessionId": session_id,
                "update": session_update.model_dump(exclude_none=True),
            },
        )

        sys.stdout.write(json.dumps(notification.model_dump(exclude_none=True)) + "\n")
        sys.stdout.flush()

        return {"stopReason": "end_turn"}

    def run(self):
        for line in sys.stdin:
            message = json.loads(line.strip())
            method = message.get("method")
            msg_id = message.get("id")

            try:
                # Route to handler using method constants
                if method == METHOD_INITIALIZE:
                    result = self.handle_initialize(message.get("params", {}))
                elif method == METHOD_SESSION_NEW:
                    result = self.handle_session_new(message.get("params", {}))
                elif method == METHOD_SESSION_PROMPT:
                    result = self.handle_session_prompt(message.get("params", {}))
                else:
                    raise Exception(f"Unknown method: {method}")

                # Use library helper to create response
                response = create_response(id=msg_id, result=result)
            except Exception as e:
                # Use library helper for error responses
                response = create_error_response(id=msg_id, code=-32603, message=str(e))

            sys.stdout.write(json.dumps(response.model_dump(exclude_none=True)) + "\n")
            sys.stdout.flush()

if __name__ == "__main__":
    EchoAgent().run()

Note: This demonstrates using the library's protocol helpers (create_response, create_notification, TextContent, etc.) instead of manual JSON construction. See examples/echo_agent.py for the complete implementation.

Example 2: Client with Session Updates

Capture and handle streaming updates from agent:

"""client_with_updates.py - Capture session/update notifications"""
import asyncio
import uuid

import anyio

from chuk_acp import (
    stdio_transport,
    send_initialize,
    send_session_new,
    ClientInfo,
    ClientCapabilities,
    TextContent,
)
from chuk_acp.protocol import (
    create_request,
    JSONRPCNotification,
    JSONRPCResponse,
    METHOD_SESSION_PROMPT,
    METHOD_SESSION_UPDATE,
)

async def main():
    async with stdio_transport("python", ["examples/echo_agent.py"]) as (read, write):
        # Initialize
        init_result = await send_initialize(
            read, write,
            protocol_version=1,
            client_info=ClientInfo(name="client", version="1.0.0"),
            capabilities=ClientCapabilities()
        )
        print(f"Connected to {init_result.agentInfo.name}")

        # Create session
        session = await send_session_new(read, write, cwd="/tmp")

        # Send prompt and capture notifications
        prompt_text = "Write a hello world function"
        print(f"User: {prompt_text}")

        request_id = str(uuid.uuid4())
        request = create_request(
            method=METHOD_SESSION_PROMPT,
            params={
                "sessionId": session.sessionId,
                "prompt": [TextContent(text=prompt_text).model_dump(exclude_none=True)],
            },
            id=request_id,
        )
        await write.send(request)

        # Collect notifications and response
        agent_messages = []
        stop_reason = None

        with anyio.fail_after(60.0):
            while stop_reason is None:
                message = await read.receive()

                # Handle session/update notifications
                if isinstance(message, JSONRPCNotification):
                    if message.method == METHOD_SESSION_UPDATE:
                        params = message.params or {}

                        # Agent message chunks
                        update = params.get("update", {})
                        if update.get("sessionUpdate") == "agent_message_chunk":
                            content = update.get("content", {})
                            if isinstance(content, dict) and "text" in content:
                                agent_messages.append(content["text"])

                        # Thoughts (optional)
                        if "thought" in params:
                            print(f"[Thinking: {params['thought']}]")

                        # Tool calls (optional)
                        if "toolCall" in params:
                            tool = params["toolCall"]
                            print(f"[Calling: {tool.get('name')}]")

                # Handle response
                elif isinstance(message, JSONRPCResponse):
                    if message.id == request_id:
                        result = message.result
                        if isinstance(result, dict):
                            stop_reason = result.get("stopReason")

        # Display captured agent messages
        if agent_messages:
            print(f"Agent: {''.join(agent_messages)}")

        print(f"Completed: {stop_reason}")

asyncio.run(main())

Key Point: To capture session/update notifications, you need to manually handle the request/response loop instead of using send_session_prompt(), which discards notifications. See examples/simple_client.py for a complete working example.

Example 3: Agent with File System Access

Agent that can read/write files:

"""file_agent.py - Agent with filesystem capabilities"""
from chuk_acp.protocol.types import AgentCapabilities

# Declare filesystem capabilities
capabilities = AgentCapabilities(
    filesystem=True  # Enables fs/read_text_file and fs/write_text_file
)

async def handle_file_operation(session_id: str, operation: str, path: str):
    """Request file access from client."""

    # Request permission
    permission = await send_session_request_permission(
        read, write,
        session_id=session_id,
        request=PermissionRequest(
            id="perm-123",
            description=f"Read file: {path}",
            tools=[{"name": "fs/read_text_file", "arguments": {"path": path}}]
        )
    )

    if permission.granted:
        # Read the file via client
        # (Client implements fs/read_text_file method)
        pass

Example 4: Multi-Session Client

Manage multiple concurrent sessions:

"""multi_session_client.py"""
import asyncio
from chuk_acp import stdio_transport, send_session_new, send_session_prompt

async def create_and_run_session(read, write, cwd: str, prompt: str):
    """Create a session and send a prompt."""
    session = await send_session_new(read, write, cwd=cwd)
    result = await send_session_prompt(
        read, write,
        session_id=session.sessionId,
        prompt=[TextContent(text=prompt)]
    )
    return result

async def main():
    async with stdio_transport("python", ["my_agent.py"]) as (read, write):
        # Initialize once
        await send_initialize(...)

        # Run multiple sessions concurrently
        tasks = [
            create_and_run_session(read, write, "/project1", "Refactor auth"),
            create_and_run_session(read, write, "/project2", "Add tests"),
            create_and_run_session(read, write, "/project3", "Fix bug #123"),
        ]

        results = await asyncio.gather(*tasks)
        print(f"Completed {len(results)} sessions")

asyncio.run(main())

API Reference

High-Level Client

The ACPClient provides the simplest way to interact with ACP agents:

Direct Usage

from chuk_acp import ACPClient

async with ACPClient("python", ["agent.py"]) as client:
    # Access agent information
    print(f"Agent: {client.agent_info.name}")
    print(f"Session: {client.current_session.sessionId}")

    # Send prompts
    result = await client.send_prompt("Hello!")
    print(result.full_message)  # Complete agent response
    print(result.stop_reason)   # Why agent stopped

    # Create new sessions
    new_session = await client.new_session(cwd="/other/path")

Configuration-Based Usage

Use standard ACP configuration format (compatible with Zed, VSCode, etc.):

from chuk_acp import ACPClient, AgentConfig, load_agent_config

# Method 1: Create config directly
config = AgentConfig(
    command="kimi",
    args=["--acp"],
    env={"DEBUG": "true"},
    cwd="/optional/path"
)

async with ACPClient.from_config(config) as client:
    result = await client.send_prompt("Hello!")

# Method 2: Load from JSON file
config = load_agent_config("~/.config/my-app/agent.json")
async with ACPClient.from_config(config) as client:
    result = await client.send_prompt("Hello!")

# Method 3: From dictionary (like editor configs)
config = AgentConfig(**{
    "command": "kimi",
    "args": ["--acp"],
    "env": {}
})
async with ACPClient.from_config(config) as client:
    result = await client.send_prompt("Hello!")

Example JSON config file:

{
  "command": "kimi",
  "args": ["--acp"],
  "env": {
    "DEBUG": "true",
    "LOG_LEVEL": "info"
  },
  "cwd": "/optional/path"
}

Key Classes:

ACPClient - Main client class
AgentConfig - Standard ACP configuration format
load_agent_config() - Load config from JSON file
PromptResult - Contains response and all notifications
SessionInfo - Session information
SessionUpdate - Individual notification from agent

Low-Level Protocol API

For fine-grained control over the protocol:

Protocol Helpers

JSON-RPC Message Helpers

Build protocol messages using library helpers:

from chuk_acp.protocol import (
    create_request,
    create_response,
    create_error_response,
    create_notification,
)

# Create a request
request = create_request(
    method="session/prompt",
    params={"sessionId": "session-1", "prompt": [...]},
    id="req-123"
)

# Create a response
response = create_response(id="req-123", result={"stopReason": "end_turn"})

# Create an error response
error = create_error_response(id="req-123", code=-32603, message="Internal error")

# Create a notification
from chuk_acp.protocol.types import SessionUpdate, TextContent

session_update = SessionUpdate(
    sessionUpdate="agent_message_chunk",
    content=TextContent(text="Hello!")
)
notification = create_notification(
    method="session/update",
    params={
        "sessionId": "session-1",
        "update": session_update.model_dump(exclude_none=True)
    }
)

Method Constants

Use constants instead of string literals for protocol methods:

from chuk_acp.protocol import (
    METHOD_INITIALIZE,
    METHOD_SESSION_NEW,
    METHOD_SESSION_PROMPT,
    METHOD_SESSION_UPDATE,
    METHOD_SESSION_CANCEL,
    METHOD_FS_READ_TEXT_FILE,
    METHOD_FS_WRITE_TEXT_FILE,
    METHOD_TERMINAL_CREATE,
    # ... and more
)

# Use in message routing
if method == METHOD_INITIALIZE:
    # Handle initialize
    pass
elif method == METHOD_SESSION_PROMPT:
    # Handle prompt
    pass

Transport

`stdio_transport(command, args)`

Create a stdio transport connection to an agent.

async with stdio_transport("python", ["agent.py"]) as (read_stream, write_stream):
    # Use streams for communication
    pass

Initialization

`send_initialize(read, write, protocol_version, client_info, capabilities)`

Initialize the connection and negotiate capabilities.

result = await send_initialize(
    read_stream,
    write_stream,
    protocol_version=1,
    client_info=ClientInfo(name="my-client", version="1.0.0"),
    capabilities=ClientCapabilities(filesystem=True)
)
# result.agentInfo, result.capabilities, result.protocolVersion

Session Management

`send_session_new(read, write, cwd, mcp_servers=None, mode=None)`

Create a new session.

session = await send_session_new(
    read_stream,
    write_stream,
    cwd="/absolute/path",
    mode="code"  # Optional: ask, architect, code
)
# session.sessionId

`send_session_prompt(read, write, session_id, prompt)`

Send a prompt to the agent.

result = await send_session_prompt(
    read_stream,
    write_stream,
    session_id="session-123",
    prompt=[
        TextContent(text="Write a function"),
        ImageContent(data="base64...", mimeType="image/png")
    ]
)
# result.stopReason: end_turn, max_tokens, cancelled, refusal

Note: send_session_prompt discards session/update notifications from the agent. To capture agent responses (message chunks, thoughts, tool calls), manually handle the request/response loop. See Example 2 or examples/simple_client.py for details.

`send_session_cancel(write, session_id)`

Cancel an ongoing prompt turn.

await send_session_cancel(write_stream, session_id="session-123")

Content Types

`TextContent(text)`

Plain text content.

content = TextContent(text="Hello, world!")

`ImageContent(data, mimeType)`

Base64-encoded image.

content = ImageContent(
    data="iVBORw0KGgoAAAANSUhEUgA...",
    mimeType="image/png"
)

`AudioContent(data, mimeType)`

Base64-encoded audio.

content = AudioContent(
    data="SUQzBAA...",
    mimeType="audio/mpeg"
)

Protocol Support

chuk-acp implements the complete ACP v1 specification.

✅ Baseline Agent Methods (Required)

Method	Description	Status
`initialize`	Protocol handshake and capability negotiation	✅
`authenticate`	Optional authentication	✅
`session/new`	Create new conversation sessions	✅
`session/prompt`	Process user prompts	✅
`session/cancel`	Cancel ongoing operations	✅

✅ Optional Agent Methods

Method	Capability	Status
`session/load`	Resume previous sessions	✅
`session/set_mode`	Change session modes	✅

✅ Client Methods (Callbacks)

Method	Description	Status
`session/request_permission`	Request user approval for actions	✅
`fs/read_text_file`	Read file contents	✅
`fs/write_text_file`	Write file contents	✅
`terminal/create`	Create terminal sessions	✅
`terminal/output`	Stream terminal output	✅
`terminal/release`	Release terminal control	✅
`terminal/wait_for_exit`	Wait for command completion	✅
`terminal/kill`	Terminate running commands	✅

✅ Content Types

Text content (baseline - always supported)
Image content (base64-encoded)
Audio content (base64-encoded)
Embedded resources
Resource links
Annotations

✅ Session Features

Session management (create, load, cancel)
Multiple parallel sessions
Session modes: ask, architect, code
Session history replay
MCP server integration

✅ Tool Integration

Tool calls with status tracking (pending, in_progress, completed, failed)
Permission requests
File location tracking
Structured output (diffs, terminals, content)
Slash commands (optional)

✅ Protocol Requirements

File paths: All paths must be absolute ✅
Line numbers: 1-based indexing ✅
JSON-RPC 2.0: Strict compliance ✅
Extensibility: _meta fields and custom methods ✅

Architecture

Project Structure

chuk-acp/
├── src/chuk_acp/
│   ├── protocol/              # Core protocol implementation
│   │   ├── jsonrpc.py         # JSON-RPC 2.0 (requests, responses, errors)
│   │   ├── acp_pydantic_base.py # Optional Pydantic support
│   │   ├── types/             # Protocol type definitions
│   │   │   ├── content.py     # Content types (text, image, audio)
│   │   │   ├── capabilities.py # Client/agent capabilities
│   │   │   ├── session.py     # Session types and modes
│   │   │   ├── tools.py       # Tool calls and permissions
│   │   │   ├── plan.py        # Task planning types
│   │   │   ├── terminal.py    # Terminal integration
│   │   │   └── ...
│   │   └── messages/          # Message handling
│   │       ├── initialize.py  # Initialize/authenticate
│   │       ├── session.py     # Session management
│   │       ├── filesystem.py  # File operations
│   │       ├── terminal.py    # Terminal operations
│   │       └── send_message.py # Core messaging utilities
│   │
│   ├── transport/             # Transport layer
│   │   ├── base.py            # Abstract transport interface
│   │   └── stdio.py           # Stdio transport (subprocess)
│   │
│   └── __init__.py            # Public API exports
│
├── examples/                  # Working examples
│   ├── echo_agent.py          # Simple echo agent
│   ├── simple_client.py       # Basic client
│   ├── quick_start.py         # Getting started
│   └── comprehensive_demo.py  # Full-featured demo
│
├── tests/                     # Test suite
│   ├── test_protocol_compliance.py  # Spec compliance
│   ├── test_jsonrpc.py        # JSON-RPC tests
│   ├── test_types.py          # Type system tests
│   ├── test_messages.py       # Message handling
│   └── test_stdio_transport.py # Transport tests
│
└── .github/                   # CI/CD workflows
    ├── workflows/
    │   ├── ci.yml             # Testing and linting
    │   ├── publish.yml        # PyPI publishing
    │   └── codeql.yml         # Security scanning
    └── ...

Design Principles

Protocol First: Strict adherence to ACP specification
Type Safety: Comprehensive type hints throughout
Optional Dependencies: Pydantic is optional, not required
Async by Default: Built on anyio for async/await
Extensibility: Custom methods and _meta fields supported
Testability: Loosely coupled, dependency injection
Zero-Config: Works out of the box with sensible defaults

Layer Separation

┌─────────────────────────────────────┐
│     User Code (Agents/Clients)      │
├─────────────────────────────────────┤
│     High-Level API (messages/)      │  ← send_initialize, send_prompt, etc.
├─────────────────────────────────────┤
│    Protocol Layer (types/, jsonrpc) │  ← Content types, capabilities
├─────────────────────────────────────┤
│    Transport Layer (transport/)     │  ← Stdio, future: WebSocket, HTTP
└─────────────────────────────────────┘

Testing

Running Tests

# Run all tests
make test

# Run with coverage
make test-cov

# Run specific test file
uv run pytest tests/test_protocol_compliance.py -v

# Test without Pydantic (fallback mode)
uv pip uninstall pydantic
uv run pytest

Test Categories

Protocol Compliance (test_protocol_compliance.py): Validates ACP spec adherence
JSON-RPC (test_jsonrpc.py): JSON-RPC 2.0 implementation
Types (test_types.py): Type system and content types
Messages (test_messages.py): Message handling and serialization
Transport (test_stdio_transport.py): Transport layer

Code Quality Checks

# Format code
make format

# Lint
make lint

# Type check
make mypy

# Security scan
make security

# All checks
make check

Relationship to MCP

ACP and MCP (Model Context Protocol) are complementary protocols:

Protocol	Purpose	Focus
MCP	What data/tools agents can access	Context & tools
ACP	Where the agent lives in your workflow	Agent lifecycle

Integration

ACP reuses MCP data structures for content types and resources:

from chuk_acp.protocol.types import (
    TextContent,      # From MCP
    ImageContent,     # From MCP
    ResourceContent,  # From MCP
)

# ACP sessions can specify MCP servers
session = await send_session_new(
    read, write,
    cwd="/project",
    mcp_servers=[
        MCPServer(
            name="filesystem",
            command="npx",
            args=["-y", "@modelcontextprotocol/server-filesystem", "/path"]
        )
    ]
)

When to Use What

Use ACP to build AI coding agents that integrate with editors
Use MCP to provide context and tools to language models
Use both for a complete AI-powered development environment

Contributing

We welcome contributions! Please see our Contributing Guide for:

Development setup
Code style and standards
Testing requirements
Pull request process
Release workflow

Quick Start for Contributors

# Clone and setup
git clone https://github.com/chuk-ai/chuk-acp.git
cd chuk-acp
uv venv
source .venv/bin/activate
uv pip install -e ".[dev,pydantic]"

# Run checks
make check

# Run examples
cd examples && python simple_client.py

Areas for Contribution

🐛 Bug fixes and issue resolution
✨ New features (check ACP spec for ideas)
📚 Documentation improvements
🧪 Additional test coverage
🌐 Additional transports (WebSocket, HTTP, etc.)
🎨 Example agents and clients
🔧 Tooling and developer experience

License

This project is licensed under the Apache License 2.0.

See LICENSE for full details.

Project details

These details have not been verified by PyPI

Project links

Release history Release notifications | RSS feed

This version

0.3.2

Nov 9, 2025

0.3.1

Nov 9, 2025

0.3.0

Nov 9, 2025

0.2.2

Nov 9, 2025

0.2.1

Nov 9, 2025

0.2.0

Nov 9, 2025

0.1.4

Nov 9, 2025

0.1.3

Nov 8, 2025

0.1.2

Nov 8, 2025

0.1.1

Nov 8, 2025

Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

chuk_acp-0.3.2.tar.gz (174.6 kB view details)

Uploaded Nov 9, 2025 Source

Built Distribution

If you're not sure about the file name format, learn more about wheel file names.

The dropdown lists show the available interpreters, ABIs, and platforms. Enable javascript to be able to filter the list of wheel files.

chuk_acp-0.3.2-py3-none-any.whl (77.7 kB view details)

Uploaded Nov 9, 2025 Python 3

File details

Details for the file chuk_acp-0.3.2.tar.gz.

File metadata

Download URL: chuk_acp-0.3.2.tar.gz
Upload date: Nov 9, 2025
Size: 174.6 kB
Tags: Source
Uploaded using Trusted Publishing? No
Uploaded via: twine/6.1.0 CPython/3.12.2

File hashes

Hashes for chuk_acp-0.3.2.tar.gz
Algorithm	Hash digest
SHA256	`84eb54145e889d2bce19b4640d2b5aa98cf3825d4f50afd986d618f552beaf5f`
MD5	`81b59f583108708c3b4221b2682fc56f`
BLAKE2b-256	`395e4804dfd218769874c24419bd99e65e3d0792bf296e240e7591a0b4edb534`

See more details on using hashes here.

File details

Details for the file chuk_acp-0.3.2-py3-none-any.whl.

File metadata

Download URL: chuk_acp-0.3.2-py3-none-any.whl
Upload date: Nov 9, 2025
Size: 77.7 kB
Tags: Python 3
Uploaded using Trusted Publishing? No
Uploaded via: twine/6.1.0 CPython/3.12.2

File hashes

Hashes for chuk_acp-0.3.2-py3-none-any.whl
Algorithm	Hash digest
SHA256	`cf6ac9a9361e5fd86bafcf07307cd83c3a54df71fa38645b2edbfb8294cebaea`
MD5	`5d1cc2d16d68149c737dd680314a7b66`
BLAKE2b-256	`71306713b065268996891291ff6ddd43a9c4558790308ea6d65274a9aa442b94`

See more details on using hashes here.

chuk-acp 0.3.2

Navigation

Verified details

Maintainers

Unverified details

Project links

Meta

Classifiers

Project description

chuk-acp

📖 Table of Contents

Overview

Why ACP?

The Problem

The Solution

Features

🎯 Complete ACP Implementation

🔧 Developer-Friendly

🚀 Flexible & Extensible

🛡️ Quality & Security

Quick Start (60 Seconds!)

🚀 Try Without Installation

Or Connect to External Agents

Installation

Using uvx (Recommended for One-Off Usage)

Using uv (Recommended for Development)

Using pip

Development Installation

Requirements

Quick Start

CLI Tool - Interactive Chat with Any Agent

Try It Now (No Installation!)

After Installation

Interactive Mode Commands

Example Interactive Session

Configuration Files

Important: Using -- for Agent Flags

Using with Editors (Zed, VSCode, etc.)

Zed Configuration

The Easiest Way: ACPClient

Building an Agent

🚀 Even Easier: chuk-acp-agent

More Examples

Option B: Build Your Own (10 Minutes)

Step 1: Install

Step 2: Create an Agent

Step 3: Create a Client

Step 4: Run It!

What You Learned

Next Steps

Core Concepts

The Agent-Client Model

Key Components

1. Protocol Layer (chuk_acp.protocol)

2. Transport Layer (chuk_acp.transport)

3. Type System (chuk_acp.protocol.types)

The ACP Flow

Complete Examples

Example 1: Echo Agent (Using Library)

Example 2: Client with Session Updates

Example 3: Agent with File System Access

Example 4: Multi-Session Client

API Reference

High-Level Client

Direct Usage

Configuration-Based Usage

Low-Level Protocol API

Protocol Helpers

JSON-RPC Message Helpers

Method Constants

Transport

stdio_transport(command, args)

Initialization

send_initialize(read, write, protocol_version, client_info, capabilities)

Session Management

send_session_new(read, write, cwd, mcp_servers=None, mode=None)

send_session_prompt(read, write, session_id, prompt)

send_session_cancel(write, session_id)

Content Types

TextContent(text)

Important: Using `--` for Agent Flags

1. Protocol Layer (`chuk_acp.protocol`)

2. Transport Layer (`chuk_acp.transport`)

3. Type System (`chuk_acp.protocol.types`)

`stdio_transport(command, args)`

`send_initialize(read, write, protocol_version, client_info, capabilities)`

`send_session_new(read, write, cwd, mcp_servers=None, mode=None)`

`send_session_prompt(read, write, session_id, prompt)`

`send_session_cancel(write, session_id)`

`TextContent(text)`

`ImageContent(data, mimeType)`

`AudioContent(data, mimeType)`