0.1.3

acp-easy

A lightweight, LangChain-style Python wrapper around the ACP SDK that makes it easy to talk to ACP-compliant agent servers, with auto agent-discovery, sync and async support, and session helpers.

Install

pip install acp-easy

Quick start

from acp_easy import AcpClient

client = AcpClient(base_url="http://localhost:8000", agent="my-agent")

reply = client.chat(
    messages=[
        {"role": "user", "content": "Hello!"},
    ],
    model="gpt-4o-mini",
)

print(reply)

Each message can be either:

• a simple {"role": ..., "content": ...} dict
• a richer dict with parts
• an acp_sdk Message object

Finding your agent name

If you don't know what agents your ACP server exposes, list them first:

from acp_easy import AcpClient

client = AcpClient(base_url="http://localhost:8000")

agents = client.list_agents()
print(agents)
# [{'name': 'my-agent', 'description': 'No description provided.'}]

Each entry gives you the name to pass as agent=....

• If you don't pass agent at all and the server has exactly one agent, acp-easy auto-resolves it for you.
• If the server has multiple agents and you don't specify one, you'll get a clear AcpAgentResolutionError listing all available agents so you can pick one.

You can set a default agent once at client creation instead of passing it on every call:

client = AcpClient(base_url="http://localhost:8000", agent="my-agent")

Available models

Pass any of these keys as the model argument to chat():

Sync vs Async

acp-easy gives you both sync and async versions of every network call. Use sync in plain scripts, and async inside code that already runs an event loop (FastAPI, Jupyter, asyncio apps).

# Inside a FastAPI route / Jupyter notebook / any async function:
agents = await client.list_agents_async()
reply = await client.chat_async(
    messages=[{"role": "user", "content": "Hello!"}],
    model="gpt-4o-mini",
)

Calling the sync methods (chat, list_agents) from inside a running event loop will raise a clear RuntimeError telling you to use the _async version instead.

Sessions and multi-turn conversations

acp-easy can work with ACP sessions so historical context can be loaded back from the server, including message parts that point at remote resources with content_url.

Create a local session object and pass it to chat() / chat_async():

from acp_easy import AcpClient

client = AcpClient(base_url="http://localhost:8000", agent="my-agent")
session = client.new_session()

reply = client.chat(
    messages=[{"role": "user", "content": "Remember this for later."}],
    model="gpt-4o-mini",
    session=session,
)

If you already know the session ID, you can fetch it from the server and iterate its history:

session = client.get_session("session-id-from-server")

async for message in client.load_session_history_async(session):
    print(message)

Message parts can also reference remote content directly:

reply = client.chat(
    messages=[
        {
            "role": "user",
            "parts": [
                {
                    "content_type": "text/plain",
                    "content_url": "http://resource-server/very-large-text",
                }
            ],
        }
    ],
    model="gpt-4o-mini",
)

Pass the full message history as a list if you want to keep the conversation state entirely client-side:

reply = client.chat(
    messages=[
        {"role": "user", "content": "My name is Ganaik."},
        {"role": "assistant", "content": "Nice to meet you, Ganaik!"},
        {"role": "user", "content": "What's my name?"},
    ],
    model="gpt-4o-mini",
)

Error handling

acp-easy raises AcpAgentResolutionError for:

• No agents found on the server
• Multiple agents found with none specified
• Malformed messages
• Session lookup failures

from acp_easy import AcpClient, AcpAgentResolutionError

client = AcpClient(base_url="http://localhost:8000")

try:
    reply = client.chat(
        messages=[{"role": "user", "content": "Hi"}],
        model="gpt-4o-mini",
    )
except AcpAgentResolutionError as e:
    print(f"Couldn't resolve agent: {e}")

API reference

AcpClient(base_url, agent=None, timeout=60.0, session=None)

• base_url - URL of your ACP server (e.g. "http://localhost:8000")
• agent - optional default agent name; skips auto-discovery if set
• timeout - request timeout in seconds (default 60.0)
• session - optional default ACP session or session ID to reuse across calls

client.list_agents() / await client.list_agents_async()

Returns a list of {"name": ..., "description": ...} dicts for all agents on the server.

client.get_session(session_id) / await client.get_session_async(session_id)

Fetches the ACP session descriptor from /sessions/{session_id}.

client.load_session_history(session) / await client.load_session_history_async(session)

Loads the session's message history and resolves any content_url parts by making HTTP GET requests to their resource servers.

client.chat(messages, model, agent=None, session=None, session_id=None) / await client.chat_async(messages, model, agent=None, session=None, session_id=None)

Sends a message history to the resolved agent and returns the agent's text reply.

• messages - list of message dicts or acp_sdk Message objects
• model - model key from the Available models table
• agent - optional agent name; overrides the client's default for this call only
• session - optional ACP Session object or session ID
• session_id - optional session ID shortcut when you do not already have a session object