AWS Strands

from dotenv import load_dotenv
load_dotenv()  # loads XPANDER_API_KEY, AWS_ACCESS_KEY_ID, etc. before any sdk import

from xpander_sdk import on_task, Task, Agents
from strands import Agent as StrandsAgent

@on_task
async def handler(task: Task) -> Task:
    # 1. Load the xpander agent (instructions, skills, model are all on it).
    xpander_agent = await Agents(configuration=task.configuration).aget(
        agent_id=task.agent_id,
    )

    # 2. Build the Strands Agent from xpander fields. Strands wraps a model-id
    #    string as BedrockModel(model_id=...) automatically.
    native = StrandsAgent(
        name=xpander_agent.name,
        description=xpander_agent.instructions.description,
        system_prompt=xpander_agent.instructions.full,
        tools=xpander_agent.strands_tools,
        model=xpander_agent.model_name,
    )

    # 3. Run the LLM loop with the task's user message.
    result = await native.invoke_async(task.to_message())

    # 4. Write the result back so xpander can store and display it.
    #    str(result) concatenates the text blocks from result.message.
    task.result = str(result)
    return task

from strands import Agent as StrandsAgent, tool

@tool
def search_knowledge_base(query: str, num_documents: int = 5) -> list[dict]:
    """Search the agent's linked knowledge bases. Returns top-k matching documents."""
    # knowledge_bases_retriever() returns a callable: (query, agent=None, num_documents=5, **kwargs)
    retriever = xpander_agent.knowledge_bases_retriever()
    return retriever(query=query, num_documents=num_documents)

native = StrandsAgent(
    name=xpander_agent.name,
    description=xpander_agent.instructions.description,
    system_prompt=xpander_agent.instructions.full,
    # Concatenate the auto-wired tools with the KB retriever.
    tools=[*xpander_agent.strands_tools, search_knowledge_base],
    model=xpander_agent.model_name,
)

from datetime import datetime, timezone
from xpander_sdk import on_task, Task, Agents, TaskUpdateEvent, TaskUpdateEventType
from strands import Agent as StrandsAgent

@on_task
async def handler(task: Task):
    xpander_agent = await Agents(configuration=task.configuration).aget(agent_id=task.agent_id)
    native = StrandsAgent(
        name=xpander_agent.name,
        description=xpander_agent.instructions.description,
        system_prompt=xpander_agent.instructions.full,
        tools=xpander_agent.strands_tools,
        model=xpander_agent.model_name,
    )

    final_output = ""
    # stream_async yields dict events: text deltas carry a "data" key,
    # tool calls and tool results carry their own keys.
    async for event in native.stream_async(task.to_message()):
        chunk = event.get("data") if isinstance(event, dict) else None
        if not chunk:
            continue
        final_output += chunk
        # Forward each text delta to xpander's SSE stream.
        yield TaskUpdateEvent(
            type=TaskUpdateEventType.Chunk,
            task_id=task.id,
            organization_id=task.organization_id,
            time=datetime.now(timezone.utc),
            data=chunk,
        )

    task.result = final_output
    yield TaskUpdateEvent(
        type=TaskUpdateEventType.TaskFinished,
        task_id=task.id,
        organization_id=task.organization_id,
        time=datetime.now(timezone.utc),
        data=task,
    )