Dwfb Project

代理AI-Strands

其中 Strands代理说明如何使用实现Agentic AI。Strands Agent是为构建和运行AI agent而设计的开源SDK。可以轻松利用agent功能，如计划（planning）、链接（chaining thoughts）、工具调用和Reflection。这将LLM模型与tool相关联，并利用模型的推理能力规划和运行工具。目前支持Amazon Bedrock、Anthropic和Meta的模型，有Accenture、Anthropic和Meta等公司参与。

这里使用的architecture如下所示。为了验证和实现Agent的基本行为，它以docker的形式安装在EC2中，使用ALB和Cloudront，用户可以使用streamlit测试其行为。Agent生成的图片或文档可以使用S3共享，您可以使用嵌入在EC2中的MCP server/client使用互联网搜索（Tavily）、RAG（knowledge base）AWS tools（use-aws）和AWS Document。

Strands agent包括： 代理循环因为有，所以选择并运行适当的工具，并通过reasoning重复执行所需的操作。

可以并行处理工具，如下所示：

agent = Agent(
    max_parallel_tools=4  
)

Strands Agent的使用方法

在Streamlit中运行agent

app.py用户可以选择“RAG”、“Agent”，如下所示。“Agent”使用Strands agent作为MCP，必要时可以使用tool利用RAG等。为Streamlit的UI输入和结果response 会话状态管理为。

if prompt := st.chat_input("메시지를 입력하세요."):
    with st.chat_message("user"):  
        st.markdown(prompt)
    st.session_state.messages.append({"role": "user", "content": prompt})

    with st.chat_message("assistant"):
        containers = {
            "tools": st.empty(),
            "status": st.empty(),
            "notification": [st.empty() for _ in range(1000)],
            "key": st.empty()
        }
        if mode == 'Agent':
            response, image_urls = asyncio.run(chat.run_strands_agent(
                query=prompt, 
                strands_tools=selected_strands_tools, 
                mcp_servers=selected_mcp_servers, 
                containers=containers))
        st.session_state.messages.append({
            "role": "assistant", 
            "content": response,
            "images": image_urls if image_urls else []
        })

Agent的运行

使用system prompt、model和tool信息创建agent，如下所示。

def create_agent(system_prompt, tools):
    if system_prompt==None:
        system_prompt = (
            "당신의 이름은 서연이고, 질문에 대해 친절하게 답변하는 사려깊은 인공지능 도우미입니다."
            "상황에 맞는 구체적인 세부 정보를 충분히 제공합니다." 
            "모르는 질문을 받으면 솔직히 모른다고 말합니다."
        )
    model = get_model()    
    agent = Agent(
        model=model,
        system_prompt=system_prompt,
        tools=tools,
        conversation_manager=conversation_manager
    )
    return agent

chat.py运行Agent并通过stream接收结果并显示，如下所示。此时，在event中只提取“data”，然后保存为full_response，然后显示为markdown，如下所示。

async def run_strands_agent(query, strands_tools, mcp_servers, containers):
    await strands_agent.initiate_agent(
        system_prompt=None, 
        strands_tools=strands_tools, 
        mcp_servers=mcp_servers
    )

    final_result = current = ""
    with strands_agent.mcp_manager.get_active_clients(mcp_servers) as _:
        agent_stream = strands_agent.agent.stream_async(query)
        
        async for event in agent_stream:
            text = ""            
            if "data" in event:
                text = event["data"]
                logger.info(f"[data] {text}")
                current += text

            elif "result" in event:
                final = event["result"]                
                message = final.message
                if message:
                    content = message.get("content", [])
                    result = content[0].get("text", "")
                    final_result = result
    return final_result

利用对话履历

如果您想使用对话内容继续对话，可以使用SlidingWindowConversationManager获取和利用以前的对话（如下图所示）的window_size。详细代码包括： chat.py请参阅。

from strands.agent.conversation_manager import SlidingWindowConversationManager

conversation_manager = SlidingWindowConversationManager(
    window_size=10,  
)

agent = Agent(
    model=model,
    system_prompt=system,
    tools=[    
        calculator, 
        current_time,
        use_aws    
    ],
    conversation_manager=conversation_manager
)

利用MCP

将stdio_mcp_client指定为MCPClient后，使用list_tools_sync提取tool信息，如下所示。MCP tool可与strands tool一起使用，如下所示。

from strands.tools.mcp import MCPClient
from strands_tools import calculator, current_time, use_aws

stdio_mcp_client = MCPClient(lambda: stdio_client(
    StdioServerParameters(command="uvx", args=["awslabs.aws-documentation-mcp-server@latest"])
))

with stdio_mcp_client as client:
    aws_documentation_tools = client.list_tools_sync()
    logger.info(f"aws_documentation_tools: {aws_documentation_tools}")

    tools=[    
        calculator, 
        current_time,
        use_aws
    ]

    tools.extend(aws_documentation_tools)

    agent = Agent(
        model=model,
        system_prompt=system,
        tools=tools,
        conversation_manager=conversation_manager
    )

此外，用于wikipedia搜索的MCP服务器示例如下：详细代码包括： mcpserver_wikipedia.py请参阅。

from mcp.server.fastmcp import FastMCP
import wikipedia
import logging
import sys

logging.basicConfig(
    level=logging.INFO,  # Default to INFO level
    format='%(filename)s:%(lineno)d | %(message)s',
    handlers=[
        logging.StreamHandler(sys.stderr)
    ]
)
logger = logging.getLogger("rag")

mcp = FastMCP(
    "Wikipedia",
    dependencies=["wikipedia"],
)

@mcp.tool()
def search(query: str):
    logger.info(f"Searching Wikipedia for: {query}")
    
    return wikipedia.search(query)

@mcp.tool()
def summary(query: str):
    return wikipedia.summary(query)

@mcp.tool()
def page(query: str):
    return wikipedia.page(query)

@mcp.tool()
def random():
    return wikipedia.random()

@mcp.tool()
def set_lang(lang: str):
    wikipedia.set_lang(lang)
    return f"Language set to {lang}"

if __name__ == "__main__":
    mcp.run()

动态绑定MCP Server

定义MCPClientManager以动态管理MCP服务器。add_client通过MCP服务器的name、command、args和env定义MCP Client。

class MCPClientManager:
    def __init__(self):
        self.clients: Dict[str, MCPClient] = {}
        
    def add_client(self, name: str, command: str, args: List[str], env: dict[str, str] = {}) -> None:
        """Add a new MCP client"""
        self.clients[name] = MCPClient(lambda: stdio_client(
            StdioServerParameters(
                command=command, args=args, env=env
            )
        ))
    
    def remove_client(self, name: str) -> None:
        """Remove an MCP client"""
        if name in self.clients:
            del self.clients[name]
    
    @contextmanager
    def get_active_clients(self, active_clients: List[str]):
        """Manage active clients context"""
        active_contexts = []
        for client_name in active_clients:
            if client_name in self.clients:
                active_contexts.append(self.clients[client_name])

        if active_contexts:
            with contextlib.ExitStack() as stack:
                for client in active_contexts:
                    stack.enter_context(client)
                yield
        else:
            yield

# Initialize MCP client manager
mcp_manager = MCPClientManager()

使用Streamlit实现的 app.py在中选择tool可获得mcp_tools。在创建agent时，从active client获取tool list并将其用作tools，如下所示。

tools = []
for mcp_tool in mcp_servers:
    with mcp_manager.get_active_clients([mcp_tool]) as _:
        client = mcp_manager.get_client(mcp_tool)
        mcp_servers_list = client.list_tools_sync()
        tools.extend(mcp_servers_list)

agent = create_agent(system_prompt, tools)
tool_list = get_tool_list(tools)

生成的agent使用mcp_manager运行，如下所示。

with mcp_manager.get_active_clients(mcp_tools) as _:
    agent_stream = agent.stream_async(question)
    
    tool_name = ""
    async for event in agent_stream:
        if "message" in event:
            message = event["message"]
            for content in message["content"]:                
                if "text" in content:
                    final_response = content["text"]

MCP概念

在Notion中 MCP服务器官方概念提供与相同的MCP服务器。通过以下方式创建token后，在mcp.json中设置相关信息，然后在agent中使用。

1. API集成理念连接到，选择“新建API集成”。
2. 按如下所示输入并保存。

1. 复制API Secret。

1. 在“权限Tab”下，选择相应的页面。

1. 创建secret，如下所示：

secrets_client = boto3.client("secretsmanager", region_name=region)
response = secrets_client.create_secret(
    Name="notionapikey-dwfb",
    Description="secret for notion api key",
    SecretString=json.dumps(secret_config["secret_value"])
)

生成的secret如下所示。

1. mcp_gonfig.py在中设置如下：

token = get_notion_key()
return {
    "mcpServers": {
        "notionApi": {
            "command": "npx",
            "args": ["-y", "@notionhq/notion-mcp-server"],
            "env": {
                "NOTION_TOKEN": token
            }
        }
    }
}

def get_notion_key():
    get_notion_api_secret = secretsmanager.get_secret_value(
        SecretId=f"notionapikey-{projectName}"
    )
    secret = json.loads(get_notion_api_secret['SecretString'])

    return notion_key = secret['notion_api_key']

调整输出语句以符合Streamlit

按如下方式运行Agent以获取agent_stream。

with mcp_manager.get_active_clients(mcp_servers) as _:
    agent_stream = agent.stream_async(question)

为了获得用户体验，必须以stream的形式获得输出。它从agent_stream中取出event，然后从“data”中提取，并将stream结果收集到current_response中，如下所示：

async for event in agent_stream:
    if "data" in event:
        text_data = event["data"]
        current_response += text_data

        containers["notification"][index].markdown(current_response)

Strands agent通过multi step reasoning多次得出结果。要获得最终结果，请从message的content中提取text，并只提取最后一个。另外，由于每个tool的reference都不同，因此从tool content的text中提取reference，如下所示。

if "message" in event:
    message = event["message"]
    for msg_content in message["content"]:                
        result = msg_content["text"]
        current_response = ""

        tool_content = msg_content["toolResult"]["content"]
        for content in tool_content:
            content, urls, refs = get_tool_info(tool_name, content["text"])
            if refs:
                for r in refs:
                    references.append(r)

名为generate_image_with_colors的工具的最终图像路径将从event_loop_metrics中提取并用作image_urls，如下所示。

if "event_loop_metrics" in event and \
    hasattr(event["event_loop_metrics"], "tool_metrics") and \
    "generate_image_with_colors" in event["event_loop_metrics"].tool_metrics:
    tool_info = event["event_loop_metrics"].tool_metrics["generate_image_with_colors"].tool
    if "input" in tool_info and "filename" in tool_info["input"]:
        fname = tool_info["input"]["filename"]
        if fname:
            url = f"{path}/{s3_image_prefix}/{parse.quote(fname)}.png"
            if url not in image_urls:
                image_urls.append(url)

分发

部署到EC2

连接到AWS console的EC2 启动实例选择。选择Launch instance，然后输入适当的名称。（例如，es）选择“Proceed without key pair”并跳过key pair。

准备好Instance后，选择\[Connet\]-\[EC2Instance Connect\]连接，如下所示。

安装python、pip、git、boto3，如下所示。

sudo yum install python3 python3-pip git docker -y
pip install boto3

对于Workshop，请复制以下格式的Credential并在EC2终端中输入：

导入git source，如下所示。

git clone https://github.com/kyopark2014/dwfb-project

使用installer.py开始安装，如下所示。

cd dwfb-project && python3 installer.py

API实现所需的credential由secret管理。因此，安装时需要输入必要的credential，请按照以下方式提前准备credential：

• 一般互联网搜索： 塔维利搜索连接并注册后发放API密钥。这个以tvly开头。
• 搜索天气： 露天气象图连接到并发放API密钥。这时price plan选择“Free”。

安装完成后，请通过以下连接到Cloudront验证操作：

连接后，选择Agent，然后选择适当的MCP tool执行所需操作，如下所示。

如果不再需要基础架构，请使用uninstaller.py将其删除。

python uninstaller.py

更新已部署的应用程序

连接到AWS console的EC2 启动实例选择名为“app-for-dwfb”的instance id，如下所示。

选择“connect”，然后选择Session Manager进行连接。

更新后，请在浏览器中再次检查，如下所示：

cd ~/dwfb-project/ && sudo ./update.sh

检查运行日志

EC2控制台在中选择名为“app-for-dwfb”的instance id，然后使用EC2的Session Manager进行连接。

首先检查当前docker container ID，如下所示。

sudo docker ps

使用container ID检查日志，如下所示。

sudo docker logs [container ID]

实际运行结果如下：

在Local中运行

为了充分利用AWS环境， 安装AWS CLI必须。如果是在EC2上部署，则不需要单独安装。在Local中安装时，请参阅以下命令。

curl "https://awscli.amazonaws.com/awscli-exe-linux-x86_64.zip" -o "awscliv2.zip" 
unzip awscliv2.zip
sudo ./aws/install

使用AWS CLI注册AWS credential，如下所示。

aws configure

curl -fsSL https://cli.kiro.dev/install | bash

使用venv配置环境可以方便地管理软件包。设置环境：

python -m venv .venv
source .venv/bin/activate

转到之后下载的github文件夹，然后安装以下所需的其他软件包：

pip install -r requirements.txt

然后使用以下命令运行streamlit：

streamlit run application/app.py

运行结果

输入“us-west-2的AWS bucket列表是？”，通过aws cli执行必要的operation，并显示以下结果。

如果您将wikipedia设置为MCP，并询问“请说明strand”，则可以使用wikipedia的search tool获得以下结果。

您可以读取特定Cloudwatch的日志，以查看日志的奇点。

参考

Strands Python示例

Strands代理SDK

Strands代理商样品

内置工具示例

介绍Strands Agents，一个开源的AI代理SDK

use_aws.py

查看Strands Agents和开源AI代理SDK

基于亚马逊基岩的药物发现剂

Strands特工-Swarm

Strands Agent流媒体演示

使用生成型AI自动执行AWS安全检查：从Q CLI到Strands Agents

利用AI Agent解决EKS应用程序和基础架构问题

与Strands Agents和AgentCore一起实施生物和制药研究助理

Strands代理商&亚马逊基岩代理商核心워크샵

使用Agentic AI实现评论管理自动化

Strands Agent Workshop（中文）

代理人工智能研讨会：人工智能基金经理

Agentic AI基金经理

Workshop-Strands SDK和AgentCore的代理AI