Strands Agents 快速上手 – dbt on Amazon Redshift 数据质量智能体

python -m venv .venv
source .venv/bin/activate  # For macOS/Linux
pip install strands-agents strands-agents-tools dbt-core dbt-redshift

tickit_analytics:
  outputs:
    dev:
      dbname: sample_data_dev
      host: redshift
      password: redshift
      port: 5439
      schema: tickit
      threads: 4
      type: redshift
      user: awsuser
  target: dev

mv dbt-mcp/.env.example dbt-mcp/.env
## 修改 DBT_PROJECT_DIR
DBT_PROJECT_DIR=/path/to/your/dbt/project 

StdioServerParameters(
    command="/path/to/.venv/bin/dbt-mcp",
    args=[],
    env={}
)

extract(minute from starttime) as event_hour

pricepaid * 0.20 as calculated_commission

firstname || lastname as full_name

from mcp import stdio_client, StdioServerParameters
from strands.tools.mcp import MCPClient
from strands import Agent
from strands.models import BedrockModel
from strands_tools import workflow, think

mcp_client = MCPClient(lambda: stdio_client(
    StdioServerParameters(
        command="/dbt-mcp/.venv/bin/dbt-mcp",
        args=[],
        env={}
    )
))

with mcp_client:
    tools = mcp_client.list_tools_sync()
    
    print(f"Available tools: {[tool.tool_name for tool in tools]}")
    
    bedrock_model = BedrockModel(
        model_id="us.amazon.nova-premier-v1:0",
        max_tokens=10000
    )
    
    SYSTEM_PROMPT = """
    You are a data quality agent for the TICKIT ticketing system.

    TICKIT Business Background:
    - Ticketing sales system with users, events, venues, sales entities
    - Key business rules: 15% commission rate, ticket price $10-$500, standardized user names, event time accurate to hour
    - Data quality KPIs: sales record completeness, price rationality, user info standardization, time data accuracy

    Your task: Execute dbt MCP tools, analyze results, and provide precise code-level fix suggestions with exact file paths and line numbers.
    """
    
    agent = Agent(
        system_prompt=SYSTEM_PROMPT, 
        tools=tools + [workflow, think],
        model=bedrock_model
    )
    
    print("Agent messages:", len(agent.messages))

    # Step 1: Execute dbt MCP tools directly (not in workflow)
    print("=== Executing dbt MCP Tools ===")
    dbt_execution = agent("""
    Please execute the following dbt MCP tools in sequence:
    
    1. Use 'build' tool to build the tickit_analytics project
    2. Use 'test' tool to run all data quality tests  
    3. Use 'list' tool to analyze model and test coverage
    4. Use 'show' tool to examine failed test details
    
    Focus on identifying failed tests and provide specific file paths and line numbers for issues.
    Report results in English.
    """)
    
    print(f"dbt MCP Execution Results: {dbt_execution}")

workflow_creation = agent.tool.workflow(
        action="create",
        workflow_id="tickit_dual_report_workflow",
        tasks=[
            {
                "task_id": "deep_analysis",
                "description": "Execute deep analysis using think tool",
                "system_prompt": "Use think tool to analyze TICKIT data quality issues from business and technical perspectives. Provide precise code-level fixes in English.",
                "priority": 4
            },
            {
                "task_id": "english_report_generation",
                "description": "Generate comprehensive English data quality report",
                "dependencies": ["deep_analysis"],
                "system_prompt": "Generate comprehensive English data quality report with exact code fixes, file paths, and line numbers based on dbt execution results and think analysis.",
                "priority": 3
            },
            {
                "task_id": "chinese_translation",
                "description": "Translate English report to Chinese",
                "dependencies": ["english_report_generation"],
                "system_prompt": "Translate the English data quality report to Chinese, maintaining all technical details, file paths, and line numbers exactly as they are. Keep code snippets and technical terms in original form.",
                "priority": 2
            }
        ]
    )

context_message = agent(f"""
    Execute the dual reporting workflow with the following dbt execution context:
    
    dbt Results: {dbt_execution}
    
    Please complete all workflow tasks:
    1. Deep analysis using think tool
    2. English report generation with precise code fixes
    3. Chinese translation maintaining technical accuracy
    
    Focus on providing exact file paths and line numbers for all code fixes.
    Generate both English and Chinese versions.
    """)