Base Async ACP¶

Base Async ACP is the foundational asynchronous model for Agentex. It gives you full control over the task lifecycle with three handlers while Agentex takes care of transport, streaming, and message delivery.

Core Characteristics¶

Three handler methods
- @acp.on_task_create – initialize state or send a welcome message
- @acp.on_task_event_send – process each incoming event/message
- @acp.on_task_cancel – cleanup when a task is cancelled
Explicit message creation – create all messages via adk.messages.create()
Asynchronous and concurrent – multiple requests can be in-flight; your code should be async-safe
State management available – use adk.state when you need persistence across events
No durability guarantees – crashes and retries are your responsibility (see Temporal for durability)

Message Flow¶

sequenceDiagram
    participant Client
    participant Agentex
    participant Agent

    Client->>Agentex: Create Task
    Agentex->>Agent: on_task_create(params)
    Agent->>Agentex: adk.messages.create(...)

    Client->>Agentex: Send Event
    Agentex->>Agent: on_task_event_send(params)
    Agent->>Agent: Process logic / update state
    Agent->>Agentex: adk.messages.create(...)

    Client->>Agentex: Cancel Task
    Agentex->>Agent: on_task_cancel(params)

Basic Implementation¶

from agentex.lib.sdk.fastacp.fastacp import FastACP
from agentex.lib.types.fastacp import AsyncACPConfig
from agentex.lib.types.acp import CreateTaskParams, SendEventParams, CancelTaskParams
from agentex.lib import adk

# Create Base Async ACP server
acp = FastACP.create(
    acp_type="async",
    config=AsyncACPConfig(type="base")
)

@acp.on_task_create
async def handle_task_create(params: CreateTaskParams) -> None:
    # Optionally initialize state, then send a message
    await adk.messages.create(
        task_id=params.task.id,
        agent_id=params.agent.id,
        content="Welcome!"
    )

@acp.on_task_event_send
async def handle_task_event_send(params: SendEventParams) -> None:
    # Process the incoming event and respond
    await adk.messages.create(
        task_id=params.task.id,
        agent_id=params.agent.id,
        content=f"You said: {params.event.content}"
    )

@acp.on_task_cancel
async def handle_task_cancel(params: CancelTaskParams) -> None:
    # Cleanup or finalization logic
    pass

Handler Parameters¶

CreateTaskParams¶

Used in @acp.on_task_create for task initialization:

Bases: BaseModel

Parameters for task/create method.

Attributes:

Name	Type	Description
`agent`	`Agent`	The agent that the task was sent to.
`task`	`Task`	The task to be created.
`params`	`dict[str, Any] \| None`	The parameters for the task as inputted by the user.
`request`	`dict[str, Any] \| None`	Additional request context including headers forwarded to this agent.

SendEventParams¶

Used in @acp.on_task_event_send for processing events:

Bases: BaseModel

Parameters for event/send method.

Attributes:

Name	Type	Description
`agent`	`Agent`	The agent that the event was sent to.
`task`	`Task`	The task that the message was sent to.
`event`	`Event`	The event that was sent to the agent.
`request`	`dict[str, Any] \| None`	Additional request context including headers forwarded to this agent.

CancelTaskParams¶

Used in @acp.on_task_cancel for cleanup:

Bases: BaseModel

Parameters for task/cancel method.

Attributes:

Name	Type	Description
`agent`	`Agent`	The agent that the task was sent to.
`task`	`Task`	The task that was cancelled.
`request`	`dict[str, Any] \| None`	Additional request context including headers forwarded to this agent.