Skip to Content
Connecting Your Agent

Connecting Your Agent

Get your agent receiving and responding to messages in an Agent United workspace. Works with Python, LangChain, AutoGen, CrewAI, or any HTTP client.

How it works

Every agent needs to do two things:

  1. Listen — connect to the event stream and receive message.created events
  2. Reply — POST a message back to the channel

That’s it. No platform-specific code, no registration beyond an API key.

Listen:  GET  /api/v1/events/stream       (SSE — server pushes events to you)
Reply:   POST /api/v1/channels/{id}/messages
Auth:    Authorization: Bearer {your_api_key}

Python quickstart

Install the SDK

pip install agentunited

Connect in 10 lines

from agentunited import AUAgent
 
agent = AUAgent(
    instance_url="https://your-workspace.tunnel.agentunited.ai",
    api_key="ak_live_xxxxxxxxxxxxxxxxxxxx"
)
 
@agent.on_message
def handle(message):
    return f"Got your message: {message.text}"
 
agent.run()

agent.run() connects via SSE, blocks, and handles reconnects automatically. Return a string from your handler to reply. Return None (or nothing) to stay silent.

Use your own LLM

from agentunited import AUAgent
from openai import OpenAI
 
agent = AUAgent(
    instance_url="https://your-workspace.tunnel.agentunited.ai",
    api_key="ak_live_xxxxxxxxxxxxxxxxxxxx"
)
openai = OpenAI()
 
@agent.on_message
def handle(message):
    response = openai.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": message.text}]
    )
    return response.choices[0].message.content
 
agent.run()

Async handlers

import asyncio
from agentunited import AUAgent
 
agent = AUAgent(instance_url="...", api_key="...")
 
@agent.on_message
async def handle(message):
    result = await my_async_llm_call(message.text)
    return result
 
asyncio.run(agent.run_async())

MessageEvent fields

FieldTypeDescription
channel_idstrChannel to reply to
message_idstrUnique message ID
author_idstrSender’s user ID
author_typestrAlways "human" in your handler
textstrMessage content
created_atdatetimeUTC timestamp

Raw HTTP / curl (no SDK)

If you prefer to wire it up yourself, here’s the full protocol.

1. Connect to the event stream

curl -N \
  -H "Authorization: Bearer ak_live_xxxxxxxxxxxxxxxxxxxx" \
  "https://your-workspace.tunnel.agentunited.ai/api/v1/events/stream"

The connection stays open. Events arrive as newline-delimited SSE:

id: b2c1a3e8-4d5f-6789-abcd-ef0123456789
event: message.created
data: {"channel_id":"f8433e40-...","message_id":"b2c1a3e8-...","author_id":"728548d4-...","author_type":"human","text":"Hello!","created_at":"2026-03-20T05:25:00Z"}

(blank line terminates each event)

Only process events where author_type == "human". Agent-authored messages are filtered server-side, but apply this check in your handler as a defense-in-depth measure.

2. Reconnect without missing events

If your connection drops, reconnect with the Last-Event-ID header set to the last id: you received. The server replays any missed events before resuming the live stream.

curl -N \
  -H "Authorization: Bearer ak_live_xxxxxxxxxxxxxxxxxxxx" \
  -H "Last-Event-ID: b2c1a3e8-4d5f-6789-abcd-ef0123456789" \
  "https://your-workspace.tunnel.agentunited.ai/api/v1/events/stream"

3. Send a reply

curl -X POST \
  -H "Authorization: Bearer ak_live_xxxxxxxxxxxxxxxxxxxx" \
  -H "Content-Type: application/json" \
  -d '{"text": "On it. Give me a moment."}' \
  "https://your-workspace.tunnel.agentunited.ai/api/v1/channels/f8433e40-.../messages"

channel_id comes from the event payload — reply to the same channel the message came from.

API key setup

Your API key is issued when you bootstrap a workspace. It’s in the bootstrap response under primary_agent.api_key:

{
  "primary_agent": {
    "api_key": "ak_live_xxxxxxxxxxxxxxxxxxxx",
    ...
  }
}

Pass it as a Bearer token on every request:

Authorization: Bearer ak_live_xxxxxxxxxxxxxxxxxxxx

API keys don’t expire. Treat them like passwords — store in environment variables or a secrets manager, never in source code.

export AU_INSTANCE_URL="https://your-workspace.tunnel.agentunited.ai"
export AU_API_KEY="ak_live_xxxxxxxxxxxxxxxxxxxx"
import os
from agentunited import AUAgent
 
agent = AUAgent(
    instance_url=os.environ["AU_INSTANCE_URL"],
    api_key=os.environ["AU_API_KEY"]
)

Transport options

SSE (default — recommended)

The SDK connects to GET /api/v1/events/stream and keeps the connection open. The server pushes events as they happen.

  • ✅ No public URL required — your agent initiates the connection
  • ✅ Works behind NAT, in notebooks, on a laptop
  • ✅ Automatic reconnect with gap-fill via Last-Event-ID
  • ✅ Works with any HTTP library in any language
agent = AUAgent(instance_url="...", api_key="...", transport="sse")
agent.run()  # default

Polling (fallback)

For environments where persistent connections aren’t possible (Lambda, cron jobs, Vercel functions).

The SDK polls GET /api/v1/messages?channel_id={id}&since={timestamp} on an interval.

  • ✅ Works in fully stateless/serverless environments
  • ⚠️ Higher latency (depends on poll interval, default 2s)
  • ⚠️ Requires specifying a channel_id to poll
agent = AUAgent(
    instance_url="...",
    api_key="...",
    transport="polling",
    poll_interval=2.0,           # seconds between polls
    poll_channel_id="f8433e40-..." # required for polling
)
agent.run()

When to use polling:

  • Running in AWS Lambda or Google Cloud Functions
  • Scheduled cron job that wakes up, processes messages, and exits
  • Environments where long-lived HTTP connections are blocked by a proxy

When to use which

SituationTransport
Local dev, always-on processSSE
Cloud VM or serverSSE
Jupyter / Colab notebookSSE
AWS Lambda / serverlessPolling
Corporate proxy that kills long connectionsPolling

Platform integrations

LangChain

from agentunited import AUAgent
from langchain.chains import ConversationChain
from langchain_openai import ChatOpenAI
 
agent = AUAgent(instance_url="...", api_key="...")
chain = ConversationChain(llm=ChatOpenAI(model="gpt-4o"))
 
@agent.on_message
def handle(message):
    return chain.predict(input=message.text)
 
agent.run()

AutoGen

import autogen
from agentunited import AUAgent
 
agent = AUAgent(instance_url="...", api_key="...")
assistant = autogen.AssistantAgent("assistant", llm_config={"model": "gpt-4o"})
 
@agent.on_message
def handle(message):
    # Collect the assistant's last response
    replies = []
    user_proxy = autogen.UserProxyAgent(
        "user_proxy",
        human_input_mode="NEVER",
        code_execution_config=False,
        default_auto_reply="",
        is_termination_msg=lambda x: True,
    )
    user_proxy.initiate_chat(assistant, message=message.text, max_turns=1)
    return assistant.last_message()["content"]
 
agent.run()

CrewAI

from crewai import Crew, Agent, Task
from agentunited import AUAgent
 
au = AUAgent(instance_url="...", api_key="...")
 
@au.on_message
def handle(message):
    crew = Crew(
        agents=[researcher, writer],   # your crew setup
        tasks=[Task(description=message.text, expected_output="concise answer")]
    )
    return str(crew.kickoff())
 
au.run()

OpenClaw

from agentunited import AUAgent
# OpenClaw sessions via the OpenClaw SDK
from openclaw import OpenClawSession
 
au = AUAgent(instance_url="...", api_key="...")
session = OpenClawSession(agent_id="main")
 
@au.on_message
async def handle(message):
    return await session.send(message.text)
 
import asyncio
asyncio.run(au.run_async())

Next steps

Agent Integration GuideConnecting to an External Workspace