This guide shows how to discover available ACP agents and execute them using different modes (synchronous, asynchronous, and streaming) through both REST API and Python SDK.

​
Discover available agents

Before running agents, you need to know what’s available. ACP servers expose agent metadata through a discovery endpoint that lists all registered agents with their capabilities. Use REST for simple interactions or Python SDK for richer, programmatic control.

​
Using REST API

Send a GET request to the /agents endpoint to list available agents:

curl http://localhost:8000/agents

​
Using Python SDK

For programmatic access, use the Python client to discover agents:

import asyncio
from acp_sdk.client import Client

async def list_agents():
    async with Client(base_url="http://localhost:8000") as client:
        async for agent in client.agents():
            print(agent)

if __name__ == "__main__":
    asyncio.run(list_agents())
For discovering agents across multiple servers, consider using a service registry or the BeeAI Platform for centralized agent discovery.

​
Run an agent

ACP supports three execution modes, each suited for different use cases:

  • Synchronous: Best for quick operations where you need immediate results
  • Asynchronous: Ideal for long-running tasks that you can check later
  • Streaming: Perfect for real-time applications requiring incremental updates

​
Message Format

All agent interactions use a standardized message format:

{
  "role": "user",
  "parts": [
    {
      "content_type": "text/plain",
      "content": "Your message here"
    }
  ]
}

Learn more about Message Stucture.

​
Synchronous execution

Synchronous execution blocks until the agent completes processing and returns the full response.

curl -X POST http://localhost:8000/runs \
  -H "Content-Type: application/json" \
  -d '{
        "agent_name": "echo",
        "input": [{"role": "user", "parts": [{"content": "Hello"}]}],
        "mode": "sync"
      }'

Using Python SDK:

import asyncio
from acp_sdk.client import Client
from acp_sdk.models import Message, MessagePart

async def run_sync():
    async with Client(base_url="http://localhost:8000") as client:
        run = await client.run_sync(
            agent="echo",
            input=[Message(parts=[MessagePart(content="Hello")])]
        )
        print(run.output)

if __name__ == "__main__":
    asyncio.run(run_sync())

​
Asynchronous execution

Asynchronous execution returns immediately with a run ID that you can use to check status and retrieve (poll) results later.

Start async execution:

curl -X POST http://localhost:8000/runs \
  -H "Content-Type: application/json" \
  -d '{
        "agent_name": "echo",
        "input": [{"role": "user", "parts": [{"content": "Hello"}]}],
        "mode": "async"
      }'

Check the status/result using the returned run_id:

curl http://localhost:8000/runs/<run_id>

Using Python SDK:

import asyncio
from acp_sdk.client import Client
from acp_sdk.models import Message, MessagePart

async def run_async():
    async with Client(base_url="http://localhost:8000") as client:
        run = await client.run_async(
            agent="echo",
            input=[Message(parts=[MessagePart(content="Hello")])]
        )
        print("Run ID:", run.run_id)

        # Later, retrieve results
        result = await client.get_run(run.run_id)
        print(result)

if __name__ == "__main__":
    asyncio.run(run_async())

​
Streaming execution

Streaming execution provides real-time updates using Server-Sent Events (SSE). Ideal for applications that need to show progress or partial results as they’re generated.

curl -N -H "Accept: text/event-stream" -X POST http://localhost:8000/runs \
  -H "Content-Type: application/json" \
  -d '{
        "agent_name": "echo",
        "input": [{"role": "user", "parts": [{"content": "Hello"}]}],
        "mode": "stream"
      }'

Using Python SDK:

import asyncio
from acp_sdk.client import Client
from acp_sdk.models import Message, MessagePart

async def run_stream():
    async with Client(base_url="http://localhost:8000") as client:
        async for event in client.run_stream(
            agent="echo",
            input=[Message(parts=[MessagePart(content="Hello")])]
        ):
            print(event)

if __name__ == "__main__":
    asyncio.run(run_stream())