xpander provides a unified API across all agent frameworks. No matter which framework you use, you get a standardized way to:
- Schedule and list tasks - consistent task management
- Consume tools - from the catalog or custom
- Manage prompts - version-controlled system prompts
- Handle memory - PostgreSQL-backed state
- Deploy containers - K8s (cloud or self-hosted)
How it works:
- Create an agent (via Workbench, SDK, or API)
- Use
Backend().aget_args() to get framework-specific config
- Initialize your framework with the returned args
- Use
@on_task to handle incoming tasks
By using @on_task, your agent automatically supports:
- Multiple interfaces - Slack, Web UI, API, Agent-to-Agent (A2A)
- Retry mechanism - automatic retries on failure
- Auto-scaling - if unavailable, xpander schedules a new instance
- Observability - built-in logging and monitoring
Framework Examples
Agno
Strands (AWS)
Google ADK
OpenAI Agents SDK
LangChain
from xpander_sdk import Task, on_task, Backend, Configuration
from agno.agent import Agent
@on_task(
Configuration(
api_key="{XPANDER_API_KEY}",
organization_id="{XPANDER_ORGANIZATION_ID}",
agent_id="{XPANDER_AGENT_ID}"
)
)
async def my_agent_handler(task: Task):
# Initialize Agno with xpander's configuration
backend = Backend(configuration=task.configuration)
agno_agent = Agent(**await backend.aget_args(task=task))
# Run the agent
result = await agno_agent.arun(
input=task.to_message(),
files=task.get_files(),
images=task.get_images()
)
task.result = result.content
return task
What’s in backend.aget_args()? Framework-specific arguments including:| Property | Description |
|---|
model | Configured AI model instance |
instructions | System prompt from Workbench |
description | Agent description |
tools | Tools from the catalog + custom @register_tool |
db | PostgreSQL connection for memory |
session_id | Task ID for conversation tracking |
from xpander_sdk import Task, on_task, Agents, Configuration
from strands import Agent
from strands.models.openai import OpenAIModel
@on_task(
Configuration(
api_key="{XPANDER_API_KEY}",
organization_id="{XPANDER_ORGANIZATION_ID}",
agent_id="{XPANDER_AGENT_ID}"
)
)
async def my_agent_handler(task: Task):
# Fetch agent configuration from xpander
xpander_agent = await Agents(configuration=task.configuration).aget()
# Initialize Strands with xpander's configuration
strands_agent = Agent(
model=OpenAIModel(
client_args={
"api_key": "{YOUR_LLM_KEY}",
},
model_id=xpander_agent.model_name
),
description=xpander_agent.instructions.description,
system_prompt=xpander_agent.instructions.instructions,
tools=xpander_agent.strands_tools,
)
# Run the agent
result = strands_agent(
prompt=task.to_message(),
invocation_state={"user_details": task.input.user.model_dump_json()} if task.input.user else None
)
task.result = result.message["content"][0]["text"]
return task
from xpander_sdk import Task, on_task, Agents, Configuration
from google.adk.models.lite_llm import LiteLlm
from google.adk.agents import Agent
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.genai import types
@on_task(
Configuration(
api_key="{XPANDER_API_KEY}",
organization_id="{XPANDER_ORGANIZATION_ID}",
agent_id="{XPANDER_AGENT_ID}"
)
)
async def my_agent_handler(task: Task):
# Fetch agent configuration from xpander
xpander_agent = await Agents(configuration=task.configuration).aget()
# Initialize ADK with xpander's configuration
adk_agent = Agent(
name=xpander_agent.sanitized_name,
model=LiteLlm(model=f"{xpander_agent.model_provider}/{xpander_agent.model_name}"),
description=xpander_agent.instructions.description,
instruction=xpander_agent.instructions.full,
tools=xpander_agent.tools.functions,
)
session_service = InMemorySessionService()
await session_service.create_session(
app_name=xpander_agent.sanitized_name,
user_id=task.input.user.id if task.input.user else "user_id",
session_id=task.id,
)
runner = Runner(
agent=adk_agent,
app_name=xpander_agent.sanitized_name,
session_service=session_service,
)
content = types.Content(
role="user",
parts=[types.Part(text=task.to_message())],
)
final_answer = ""
# Stream events from the agent
async for event in runner.run_async(
user_id=task.input.user.id if task.input.user else "user_id",
new_message=content,
session_id=task.id
):
if event.is_final_response() and event.content and event.content.parts:
final_answer = event.content.parts[0].text
task.result = final_answer
return task
from xpander_sdk import Task, on_task, Agents, Configuration
from agents import Agent, Runner
@on_task(
Configuration(
api_key="{XPANDER_API_KEY}",
organization_id="{XPANDER_ORGANIZATION_ID}",
agent_id="{XPANDER_AGENT_ID}"
)
)
async def my_agent_handler(task: Task):
# Fetch agent configuration from xpander
xpander_agent = await Agents(configuration=task.configuration).aget()
# Initialize OpenAI Agents with xpander's configuration
agent = Agent(
name=xpander_agent.name,
instructions=xpander_agent.instructions.full,
model=xpander_agent.model_name,
tools=xpander_agent.openai_agents_sdk_tools
)
result = await Runner.run(starting_agent=agent, input=task.to_message())
task.result = result.final_output
return task
from xpander_sdk import Task, on_task, Agents, Configuration
from langchain_openai import ChatOpenAI
from langchain.agents import create_agent
from langchain_core.messages import SystemMessage
@on_task(
Configuration(
api_key="{XPANDER_API_KEY}",
organization_id="{XPANDER_ORGANIZATION_ID}",
agent_id="{XPANDER_AGENT_ID}"
)
)
async def my_agent_handler(task: Task):
# Fetch agent configuration from xpander
xpander_agent = await Agents(configuration=task.configuration).aget()
# Initialize LangChain with xpander's configuration
agent = create_agent(
model=ChatOpenAI(
model=xpander_agent.model_name,
temperature=0,
api_key="{YOUR_OPENAI_KEY}"
),
tools=xpander_agent.tools.functions
)
result = await agent.ainvoke(
input={
"messages": [
("system", xpander_agent.instructions.full),
("system", f"User details: {task.input.user.model_dump_json() if task.input.user else 'N/A'}"),
("user", task.to_message())
]
}
)
task.result = result['messages'].pop().content
return task
Local Development
Option 1: CLI
# Install the xpander CLI
npm install -g xpander-cli
# Run agent locally - connects to xpander platform for tools, prompts, and task handling
xpander agent dev [agent-name]
- Web UI -
https://<agent-name>.agents.xpander.ai
- API -
curl -X POST https://api.xpander.ai/v1/agents/<agent-id>/invoke -H 'x-api-key: <key>' -d '{"input":{"text":"hi!"}}'
- SDK -
Tasks().create(agent_id="...", input=...)
- CLI -
xpander agent invoke [agent-name] "test message"
See Invoke API for more options.
Option 2: Inline test task
Run your code directly with a test task - no CLI needed:
from xpander_sdk import on_task, Configuration
from xpander_sdk.modules.tasks.models.task import AgentExecutionInput, LocalTaskTest
test_task = LocalTaskTest(
input=AgentExecutionInput(text="Your test input here"),
)
@on_task(
Configuration(
api_key="{XPANDER_API_KEY}",
organization_id="{XPANDER_ORGANIZATION_ID}",
agent_id="{XPANDER_AGENT_ID}"
),
test_task=test_task # Runs immediately with this test input
)
async def my_agent_handler(task):
# Your agent logic here
...
Starting from scratch? Use xpander agent init [agent-name] to scaffold a new project with Dockerfile, requirements.txt, and xpander_handler.py.
Deployment
Deploy your container to xpander’s infrastructure:
# Deploy to xpander cloud
xpander deploy
# View logs
xpander agent logs [agent]
# Restart container
xpander agent restart [agent]
Next Steps
