Frameworks Guide

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]

Once running, your agent is available via:

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:

Deployment

# Deploy to xpander cloud
xpander deploy

# View logs
xpander agent logs [agent]

# Restart container
xpander agent restart [agent]

For self-hosted deployment on your own Kubernetes cluster, see the Self-Hosted Deployment Guide.

Next Steps

Events API

Learn more about the @on_task decorator and event handling

Tools API

Create custom tools with @register_tool

Examples

Explore complete agent examples

CLI Reference

Full CLI documentation

Getting Started

Build

Run

Manage

Scale

Self-Hosted

Framework Examples

Local Development

Option 1: CLI

Option 2: Inline test task

Deployment

Next Steps

Events API

Tools API

Examples

CLI Reference

Getting Started

Build

Run

Manage

Scale

Self-Hosted

​Framework Examples

​Local Development

​Option 1: CLI

​Option 2: Inline test task

​Deployment

​Next Steps

Events API

Tools API

Examples

CLI Reference

Framework Examples

Local Development

Option 1: CLI

Option 2: Inline test task

Deployment

Next Steps