Durable agent with MCP and Activity-backed tools using the Strands Agents SDK

import asyncio
import xml.etree.ElementTree as ET

from temporalio import activity

WHATS_NEW_FEED = "https://aws.amazon.com/about-aws/whats-new/recent/feed/"

@activity.defn
async def get_recent_aws_announcements(limit: int = 5) -> list[dict]:
    """Fetch the most recent AWS 'What's New' announcements from the live RSS feed.

    Use this when the user asks what is new or recently launched in AWS. Returns a
    list of {title, link, published} for the latest service launches and updates.
    """
    import requests

    response = await asyncio.to_thread(requests.get, WHATS_NEW_FEED, timeout=10)
    response.raise_for_status()

    root = ET.fromstring(response.content)
    return [
        {
            "title": item.findtext("title", ""),
            "link": item.findtext("link", ""),
            "published": item.findtext("pubDate", ""),
        }
        for item in root.findall(".//item")[:limit]
    ]

from datetime import timedelta

from temporalio import workflow
from temporalio.contrib.strands import TemporalAgent, TemporalMCPClient
from temporalio.contrib.strands.workflow import activity_as_tool

from activities.tools import get_recent_aws_announcements

SYSTEM_PROMPT = (
    "You are an AWS expert assistant. Use the AWS documentation tools to answer "
    "questions about AWS services, and the announcements tool to report recent "
    "launches. Cite documentation links when they are relevant."
)

@workflow.defn
class AWSAssistantWorkflow:
    def __init__(self) -> None:
        aws_docs = TemporalMCPClient(
            server="aws-docs",
            cache_tools=True,
            start_to_close_timeout=timedelta(seconds=60),
        )

        self.agent = TemporalAgent(
            start_to_close_timeout=timedelta(seconds=120),
            system_prompt=SYSTEM_PROMPT,
            tools=[
                aws_docs,
                activity_as_tool(
                    get_recent_aws_announcements,
                    start_to_close_timeout=timedelta(seconds=30),
                ),
            ],
        )

    @workflow.run
    async def run(self, prompt: str) -> str:
        result = await self.agent.invoke_async(prompt)
        return str(result)

import asyncio

from mcp import StdioServerParameters, stdio_client
from strands.tools.mcp import MCPClient
from temporalio.client import Client
from temporalio.contrib.strands import StrandsPlugin
from temporalio.worker import Worker

from activities.tools import get_recent_aws_announcements
from workflows.aws_assistant_workflow import AWSAssistantWorkflow

TASK_QUEUE = "strands-aws-assistant-task-queue"

def make_aws_docs_client() -> MCPClient:
    """Factory for the AWS Documentation MCP server, run locally via uvx."""
    return MCPClient(
        lambda: stdio_client(
            StdioServerParameters(
                command="uvx",
                args=["awslabs.aws-documentation-mcp-server@latest"],
            )
        )
    )

async def main():
    plugin = StrandsPlugin(mcp_clients={"aws-docs": make_aws_docs_client})
    client = await Client.connect("localhost:7233", plugins=[plugin])

    worker = Worker(
        client,
        task_queue=TASK_QUEUE,
        workflows=[AWSAssistantWorkflow],
        activities=[get_recent_aws_announcements],
    )
    print(f"Worker started, task queue: {TASK_QUEUE}")
    await worker.run()

if __name__ == "__main__":
    asyncio.run(main())

import asyncio

from temporalio.client import Client
from temporalio.common import WorkflowIDConflictPolicy
from temporalio.contrib.strands import StrandsPlugin

from workflows.aws_assistant_workflow import AWSAssistantWorkflow

TASK_QUEUE = "strands-aws-assistant-task-queue"

async def main():
    client = await Client.connect("localhost:7233", plugins=[StrandsPlugin()])

    user_input = input("Ask the AWS assistant a question: ")

    result = await client.execute_workflow(
        AWSAssistantWorkflow.run,
        user_input,
        id="strands-aws-assistant",
        task_queue=TASK_QUEUE,
        id_conflict_policy=WorkflowIDConflictPolicy.TERMINATE_EXISTING,
    )
    print(f"Result: {result}")

if __name__ == "__main__":
    asyncio.run(main())