Discover & Run Agent

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

{
  "agents": [
    { "name": "echo", "description": "Echoes everything", "metadata": {} }
  ]
}

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"
      }'

{
  "agent_name": "echo",
  "status": "completed",
  "output": [
    { 
      "role": "agent/echo",
      "parts": [{ "content_type": "text/plain", "content": "Hello" }] 
    }
  ]
}

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"
      }'

{
  "agent_name": "echo",
  "status": "created",
  "run_id": "12345678-abcd-1234-abcd-123456789abc"
}

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.run_status(run_id=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())

Introduction

Core Concepts

How To

REST API

Integrations

About

Discover available agents

Using REST API

Using Python SDK

Run an agent

Message Format

Synchronous execution

Asynchronous execution

Streaming execution

​Discover available agents

​Using REST API

​Using Python SDK

​Run an agent

​Message Format

​Synchronous execution

​Asynchronous execution

​Streaming execution

Discover available agents

Using REST API

Using Python SDK

Run an agent

Message Format

Synchronous execution

Asynchronous execution

Streaming execution