Skip to main content

​
Sync Agent Guide

This guide demonstrates how to interact with synchronous (sync) AgentEx agents using the AgentEx SDK.

​
Overview

Sync agents use a sync ACP architecture that:
  • Uses direct message-based communication via RPC
  • Returns responses with the request (synchronously)
  • Supports both streaming and non-streaming responses

​
Key API Method: send_message(), send_message_stream()

For sync agents, you always use the send_message() or send_message_stream() method to communicate with the agent. This is different from agentic agents, which use send_event() for task-based communication. 
# Sync agents use send_message() - no task creation needed!
response = client.agents.send_message(
    agent_name=AGENT_NAME,
    params={
        "content": {"type": "text", "author": "user", "content": "Hello!"}
    }
)
Key Differences:
  • No task creation required - You can send messages directly to the agent by name
  • No event creation required - You can send messages directly to the agent by name

​
Setup

​
Prerequisites

pip install agentex-sdk

​
Environment Variables

export AGENTEX_API_KEY="your-api-key"
export AGENTEX_BASE_URL="http://agentex.agentex.azure.workspace.egp.scale.com"
export SGP_ACCOUNT_ID="your-account-id"

​
Initialize the Client

from agentex import Agentex
from agentex.types import TextContent

# Initialize the AgentEx client
client = Agentex(
    base_url=os.environ.get("AGENTEX_BASE_URL"),
    api_key="your-api-key",
    default_headers={
        "x-api-key": os.environ.get("AGENTEX_API_KEY"),
        "x-selected-account-id": os.environ.get("SGP_ACCOUNT_ID"),
    },
)

AGENT_NAME = "<your-agent-name>"  # Replace with your agent name

​
Basic Usage

​
Non-Streaming Response

Send a message and receive a complete response immediately as part of the response: 
# Send a message without streaming
rpc_response = client.agents.send_message(
    agent_name=AGENT_NAME,
    params={
        "content": {
            "type": "text",
            "author": "user",
            "content": "Hello! Can you explain what machine learning is?",
        },
        "stream": False,
    },
)

# Extract and print the text content from the response
print("User: Hello! Can you explain what machine learning is?\n")
print("Agent:")
for task_message in rpc_response.result:
    content = task_message.content
    if isinstance(content, TextContent):
        print(content.content)
Output: 
User: Hello! Can you explain what machine learning is?

Agent:
Machine learning is a subset of artificial intelligence that enables systems to learn and improve from experience without being explicitly programmed...

​
Streaming Responses

For real-time responses, use streaming mode: 
from agentex.types import TextDelta
from agentex.types.agent_rpc_result import (
    StreamTaskMessageDelta,
    StreamTaskMessageDone,
    StreamTaskMessageFull
)

print("User: Can you give me a simple example of supervised learning?\n")
print("Agent: ", end="", flush=True)

# Stream the response
for chunk in client.agents.send_message_stream(
    agent_name=AGENT_NAME,
    params={
        "content": {
            "type": "text",
            "author": "user",
            "content": "Can you give me a simple example of supervised learning?",
        },
    },
):
    # The result is a TaskMessageUpdate
    task_message_update = chunk.result

    # Print text deltas as they arrive
    if isinstance(task_message_update, StreamTaskMessageDelta):
        delta = task_message_update.delta
        if isinstance(delta, TextDelta):
            print(delta.text_delta, end="", flush=True)
    elif isinstance(task_message_update, StreamTaskMessageFull):
        content = task_message_update.content
        if isinstance(content, TextContent):
            print(content.content)

print("\n")
⌘I