Giới thiệu
Trong bài viết này, chúng ta sẽ khám phá cách sử dụng Strands Agents kết hợp với các máy chủ MCP để giải quyết các vấn đề tự động hóa trong thực tế. Chúng ta sẽ đi qua các ví dụ thực tế nhằm minh chứng sức mạnh của phương pháp này trong việc tự động hóa quy trình phức tạp.
Mục lục
- Ví dụ 1: Hệ thống Tạo Dẫn và Quản lý CRM thông minh
- Ví dụ 2: Hệ thống Phản ứng Sự cố DevOps thông minh
- Ví dụ 3: Tự động hóa Tiếp thị Nội dung thông minh
- Ví dụ 4: Tối ưu hóa Cơ sở hạ tầng AWS
- Lợi ích của Phương pháp Này
- Chạy Các Ví dụ
Ví dụ 1: Hệ thống Tạo Dẫn và Quản lý CRM thông minh
Ví dụ này mô tả một hệ thống tạo dẫn tinh vi kết hợp nghiên cứu web, cập nhật CRM và tích hợp GitHub để tự động hóa quy trình bán hàng.
Cài đặt và Cấu hình
python
# Cài đặt các gói cần thiết
pip install strands-agents strands-agents-tools
pip install tavily-python firecrawl-py hubspot-api-client
pip install python-github-api
# Thiết lập biến môi trường
export TAVILY_API_KEY="your_tavily_key"
export FIRECRAWL_API_KEY="your_firecrawl_key"
export HUBSPOT_ACCESS_TOKEN="your_hubspot_token"
export GITHUB_TOKEN="your_github_token"
export OPENAI_API_KEY="your_openai_key"Cấu hình máy chủ MCP
Đầu tiên, chúng ta sẽ thiết lập các máy chủ MCP mà chúng ta sẽ sử dụng:
python
# mcp_servers_config.py
import os
from mcp.client.stdio import stdio_client, StdioServerParameters
from mcp.client.sse import sse_client
from strands.tools.mcp import MCPClient
# Máy chủ Tavily
def create_tavily_mcp_client():
return MCPClient(lambda: sse_client(
f"https://mcp.tavily.com/mcp/?tavilyApiKey={os.getenv('TAVILY_API_KEY')}"
))
# Máy chủ Firecrawl
def create_firecrawl_mcp_client():
return MCPClient(lambda: sse_client(
f"https://mcp.firecrawl.dev/{os.getenv('FIRECRAWL_API_KEY')}/v2/sse"
))
# Máy chủ HubSpot
def create_hubspot_mcp_client():
server_params = StdioServerParameters(
command="npx",
args=["-y", "hubspot-mcp-server"],
env={"HUBSPOT_ACCESS_TOKEN": os.getenv("HUBSPOT_ACCESS_TOKEN")}
)
return MCPClient(lambda: stdio_client(server_params))Hệ thống Tạo Dẫn Chính
python
# lead_generation_system.py
import asyncio
import json
from datetime import datetime
from strands import Agent
from strands.models.openai import OpenAIModel
from mcp_servers_config import (
create_tavily_mcp_client,
create_firecrawl_mcp_client,
create_hubspot_mcp_client
)
class IntelligentLeadGenerationSystem:
def __init__(self):
self.model = OpenAIModel(
api_key=os.getenv("OPENAI_API_KEY"),
model="gpt-4-turbo",
temperature=0.7
)
self.tavily_client = create_tavily_mcp_client()
self.firecrawl_client = create_firecrawl_mcp_client()
self.hubspot_client = create_hubspot_mcp_client()
async def execute_lead_generation_workflow(self, industry: str, company_size: str, location: str):
with (self.tavily_client, self.firecrawl_client, self.hubspot_client):
tavily_tools = self.tavily_client.list_tools_sync()
firecrawl_tools = self.firecrawl_client.list_tools_sync()
hubspot_tools = self.hubspot_client.list_tools_sync()
all_tools = tavily_tools + firecrawl_tools + hubspot_tools
research_agent = self.create_research_agent(all_tools)
crm_agent = self.create_crm_agent(all_tools)
results = await self.run_workflow(research_agent, crm_agent, industry, company_size, location)
return results
def create_research_agent(self, tools):
system_prompt = """
Bạn là một tác nhân nghiên cứu dẫn chuyên nghiệp.
"""
return Agent(
model=self.model,
tools=tools,
system_prompt=system_prompt
)
def create_crm_agent(self, tools):
system_prompt = """
Bạn là một chuyên gia quản lý CRM.
"""
return Agent(
model=self.model,
tools=tools,
system_prompt=system_prompt
)
async def run_workflow(self, research_agent, crm_agent, industry, company_size, location):
research_query = f"""
Tôi cần tìm kiếm các dẫn chất lượng cao cho giải pháp phần mềm B2B của chúng tôi.
"""
research_results = research_agent(research_query)
crm_query = f"""
Dựa trên các dẫn đã được xác định, vui lòng tạo hồ sơ công ty trong HubSpot.
"""
crm_results = crm_agent(crm_query)
return {
"research_results": research_results.message,
"crm_results": crm_results.message,
"timestamp": datetime.now().isoformat()
}
# Ví dụ sử dụng
async def main():
system = IntelligentLeadGenerationSystem()
results = await system.execute_lead_generation_workflow(
industry="Công nghệ Tài chính",
company_size="50-200 nhân viên",
location="Khu vực Vịnh San Francisco"
)
print(json.dumps(results, indent=2))
if __name__ == "__main__":
asyncio.run(main())Ví dụ 2: Hệ thống Phản ứng Sự cố DevOps thông minh
Ví dụ này mô tả một hệ thống phản ứng sự cố tự động kết hợp giám sát AWS, tạo vấn đề GitHub và giải quyết vấn đề có cấu trúc.
python
# devops_incident_response.py
import os
import json
from datetime import datetime
from strands import Agent
from strands.models.anthropic import AnthropicModel
from mcp_servers_config import (
create_sequential_thinking_mcp_client,
create_github_mcp_client
)
class DevOpsIncidentResponseSystem:
def __init__(self):
self.model = AnthropicModel(
api_key=os.getenv("ANTHROPIC_API_KEY"),
model="claude-3-5-sonnet-20241022"
)
self.sequential_client = create_sequential_thinking_mcp_client()
self.github_client = create_github_mcp_client()
async def handle_incident(self, alert_data: dict):
with (self.sequential_client, self.github_client):
incident_agent = Agent(
model=self.model,
tools=[self.sequential_client, self.github_client],
system_prompt="""
Bạn là một chuyên gia phản ứng sự cố DevOps.
"""
)
response = incident_agent(f"{json.dumps(alert_data)}")
return response.message
async def main():
system = DevOpsIncidentResponseSystem()
alert_data = {"incident_id": "INC-2024-001", "severity": "high"}
result = await system.handle_incident(alert_data)
print(result)
if __name__ == "__main__":
asyncio.run(main())Ví dụ 3: Tự động hóa Tiếp thị Nội dung thông minh
Ví dụ này mô tả tự động hóa tiếp thị nội dung kết hợp nghiên cứu web, phân tích nội dung và quản lý mạng xã hội.
python
# content_marketing_automation.py
import os
from strands import Agent
from strands.models.openai import OpenAIModel
from mcp_servers_config import (
create_tavily_mcp_client,
create_firecrawl_mcp_client,
create_hubspot_mcp_client
)
class ContentMarketingAutomationSystem:
def __init__(self):
self.model = OpenAIModel(
api_key=os.getenv("OPENAI_API_KEY"),
model="gpt-4-turbo",
temperature=0.8
)
self.tavily_client = create_tavily_mcp_client()
self.firecrawl_client = create_firecrawl_mcp_client()
self.hubspot_client = create_hubspot_mcp_client()
async def create_content_campaign(self, topic: str, target_audience: str, content_types: list):
with (self.tavily_client, self.firecrawl_client, self.hubspot_client):
content_strategist = Agent(
model=self.model,
tools=[self.tavily_client, self.firecrawl_client, self.hubspot_client],
system_prompt="""
Bạn là một chuyên gia chiến lược tiếp thị nội dung.
"""
)
campaign_query = f"""
Tôi cần tạo một chiến dịch tiếp thị nội dung.
"""
result = content_strategist(campaign_query)
return result.message
async def main():
system = ContentMarketingAutomationSystem()
result = await system.create_content_campaign(
topic="Tự động hóa Doanh nghiệp bằng AI",
target_audience="Chủ doanh nghiệp nhỏ",
content_types=["bài viết blog", "nội dung mạng xã hội"]
)
print(result)
if __name__ == "__main__":
asyncio.run(main())Ví dụ 4: Tối ưu hóa Cơ sở hạ tầng AWS
python
# aws_infrastructure_optimization.py
import os
from strands import Agent
from strands.models.anthropic import AnthropicModel
from mcp_servers_config import create_sequential_thinking_mcp_client
class AWSInfrastructureOptimizer:
def __init__(self):
self.model = AnthropicModel(
api_key=os.getenv("ANTHROPIC_API_KEY"),
model="claude-3-5-sonnet-20241022"
)
self.sequential_client = create_sequential_thinking_mcp_client()
async def optimize_infrastructure(self, environment: str):
optimizer_agent = Agent(
model=self.model,
tools=[self.sequential_client],
system_prompt="""
Bạn là một chuyên gia tối ưu hóa cơ sở hạ tầng AWS.
"""
)
optimization_query = f"""
Vui lòng phân tích và tối ưu hóa cơ sở hạ tầng AWS cho môi trường {environment}.
"""
result = optimizer_agent(optimization_query)
return result.message
async def main():
optimizer = AWSInfrastructureOptimizer()
result = await optimizer.optimize_infrastructure("production")
print(result)
if __name__ == "__main__":
asyncio.run(main())Lợi ích của Phương pháp Này
- Trí Tuệ Mô Hình Trước: Phương pháp mô hình trước cho phép các tác nhân linh hoạt thích ứng chiến lược dựa trên tình huống cụ thể.
- Tích Hợp MCP Liền Mạch: Các máy chủ MCP cho phép các tác nhân khám phá và sử dụng công cụ một cách linh hoạt.
- Kiến Trúc Sẵn Sàng Sản Xuất: Các tính năng quan sát, xử lý lỗi và khả năng mở rộng được tích hợp sẵn.
- Điều Phối Đa Tác Nhân: Các tác nhân chuyên môn có thể hợp tác trong các nhiệm vụ phức tạp.
Chạy Các Ví dụ
- Cài đặt các phụ thuộc:
bash
pip install strands-agents strands-agents-tools
pip install tavily-python firecrawl-py hubspot-api-client- Cấu hình các máy chủ MCP:
bash
npx -y @modelcontextprotocol/server-sequential-thinking
npx -y hubspot-mcp-server
npx -y github-mcp-server- Thiết lập biến môi trường:
bash
export TAVILY_API_KEY="your_key"
export FIRECRAWL_API_KEY="your_key"
export HUBSPOT_ACCESS_TOKEN="your_token"
export GITHUB_TOKEN="your_token"
export OPENAI_API_KEY="your_key"- Chạy các ví dụ:
bash
python lead_generation_system.py
python devops_incident_response.py
python content_marketing_automation.py
python aws_infrastructure_optimization.pyKết luận
Bài viết đã trình bày cách Strands Agents kết hợp với các máy chủ MCP có thể tạo ra các hệ thống tự động hóa thông minh, ứng dụng cho nhiều tình huống trong thực tế. Hãy thử nghiệm và khám phá sức mạnh của tự động hóa để tối ưu hóa quy trình làm việc của bạn ngay hôm nay! Nếu bạn thấy bài viết hữu ích, hãy theo dõi để cập nhật thêm nhiều nội dung khác nhé!
