Sync Agent

# 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!"}
    }
)

pip install agentex-sdk

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"

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

# 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)

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...

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")