Strands Agents 완전 가이드 2편 — 실전 튜토리얼

1편에서 Strands Agents의 개요와 철학을 다뤘어요.

이번 편은 실제로 손으로 만들어봐요.

이번 편에서 만드는 것:
1. 설치 및 환경 설정
2. 첫 번째 에이전트 (5분)
3. 커스텀 툴 만들기
4. Anthropic API 직접 연결 (AWS 없이)
5. 멀티에이전트 구성
6. MCP 서버 연결
7. Lambda 배포

 

---

1단계 — 설치

# Python 3.10 이상 필요
python --version

# 가상환경 생성
python -m venv strands-env
source strands-env/bin/activate   # Mac/Linux
# strands-env\Scripts\activate    # Windows

# 기본 설치
pip install strands-agents strands-agents-tools

# MCP 지원 추가
pip install strands-agents[mcp]

---

2단계 — 모델 설정

Strands는 기본적으로 Amazon Bedrock을 쓰지만, AWS 없이도 Anthropic API 직접 연결 가능해요.

방법 A — Anthropic API 직접 (AWS 없이)

export ANTHROPIC_API_KEY=sk-ant-xxxxx

from strands import Agent
from strands.models import AnthropicModel

model = AnthropicModel(
    client_params={"api_key": "sk-ant-xxxxx"},
    model_id="claude-sonnet-4-6",
)

agent = Agent(model=model)
agent("안녕하세요!")

방법 B — Amazon Bedrock (AWS 사용)

# AWS CLI 설치 후
aws configure
# AWS Access Key ID: xxxxx
# AWS Secret Access Key: xxxxx
# Default region: us-west-2

# Bedrock에서 Claude 모델 활성화 필요:
# AWS 콘솔 → Bedrock → Model Access
# → Anthropic Claude 모델 활성화

from strands import Agent
from strands.models.bedrock import BedrockModel

model = BedrockModel(
    model_id="us.anthropic.claude-sonnet-4-20250514",
    region_name="us-west-2",
)

agent = Agent(model=model)
agent("안녕하세요!")

방법 C — OpenAI

from strands import Agent
from strands.models import OpenAIModel

model = OpenAIModel(
    client_params={"api_key": "sk-xxxxx"},
    model_id="gpt-5.4-mini",
)

agent = Agent(model=model)

방법 D — Ollama (로컬, 완전 무료)

# Ollama 설치 후
ollama pull llama4:scout

from strands import Agent
from strands.models import OllamaModel

model = OllamaModel(model_id="llama4:scout")
agent = Agent(model=model)

---

3단계 — 첫 번째 에이전트

가장 간단한 에이전트

from strands import Agent

# 모델 설정 없으면 환경변수에서 자동으로 읽음
agent = Agent()

# 실행
response = agent("파이썬으로 피보나치 수열 10개 출력해줘")
print(response)

내장 툴 사용

from strands import Agent
from strands_tools import calculator, http_request, file_read, file_write

agent = Agent(
    system_prompt="당신은 유용한 AI 어시스턴트입니다.",
    tools=[calculator, http_request, file_read, file_write]
)

# 계산
agent("1764의 제곱근이 얼마야?")

# 웹 요청
agent("https://api.github.com/repos/strands-agents/sdk-python 정보 가져와줘")

# 파일 작업
agent("hello.txt 파일에 '안녕하세요' 라고 써줘")

대화 이어가기

from strands import Agent

agent = Agent()

# 첫 번째 메시지
agent("내 이름은 셀이야")

# 이전 대화 기억
agent("내 이름이 뭐야?")
# → "셀이라고 하셨네요!"

대화 히스토리가 자동으로 유지돼요.

---

4단계 — 커스텀 툴 만들기

기본 툴

from strands import Agent, tool

@tool
def get_current_time() -> str:
    """현재 시간을 반환합니다."""
    from datetime import datetime
    return datetime.now().strftime("%Y년 %m월 %d일 %H시 %M분")

@tool
def calculate_bmi(weight_kg: float, height_m: float) -> str:
    """BMI를 계산합니다.

    Args:
        weight_kg: 몸무게 (킬로그램)
        height_m: 키 (미터)

    Returns:
        BMI 값과 판정 결과
    """
    bmi = weight_kg / (height_m ** 2)

    if bmi < 18.5:
        status = "저체중"
    elif bmi < 23:
        status = "정상"
    elif bmi < 25:
        status = "과체중"
    else:
        status = "비만"

    return f"BMI: {bmi:.1f} ({status})"

# 에이전트에 등록
agent = Agent(
    system_prompt="건강 관련 질문에 답하는 어시스턴트입니다.",
    tools=[get_current_time, calculate_bmi]
)

agent("키 175cm, 몸무게 70kg인 사람의 BMI는?")

외부 API 연동 툴

from strands import Agent, tool
import requests

@tool
def get_github_repo_info(repo: str) -> dict:
    """GitHub 레포지토리 정보를 가져옵니다.

    Args:
        repo: 레포지토리 경로 (예: 'strands-agents/sdk-python')

    Returns:
        레포지토리 정보 딕셔너리
    """
    response = requests.get(f"https://api.github.com/repos/{repo}")
    data = response.json()

    return {
        "이름": data["name"],
        "설명": data["description"],
        "스타": data["stargazers_count"],
        "언어": data["language"],
        "최근 업데이트": data["updated_at"][:10],
    }

@tool
def search_pypi(package_name: str) -> dict:
    """PyPI에서 패키지 정보를 검색합니다.

    Args:
        package_name: 패키지 이름

    Returns:
        패키지 정보
    """
    response = requests.get(f"https://pypi.org/pypi/{package_name}/json")
    data = response.json()["info"]

    return {
        "이름": data["name"],
        "버전": data["version"],
        "설명": data["summary"],
        "작성자": data["author"],
    }

agent = Agent(
    system_prompt="Python 패키지와 GitHub 레포 정보를 제공하는 어시스턴트입니다.",
    tools=[get_github_repo_info, search_pypi]
)

agent("strands-agents PyPI 패키지 정보랑 GitHub 레포 정보 비교해줘")

AWS 서비스 연동 툴

import boto3
from strands import Agent, tool

@tool
def read_s3_file(bucket: str, key: str) -> str:
    """S3에서 파일 내용을 읽습니다.

    Args:
        bucket: S3 버킷 이름
        key: 파일 경로 (예: 'docs/readme.txt')

    Returns:
        파일 내용 문자열
    """
    s3 = boto3.client("s3")
    response = s3.get_object(Bucket=bucket, Key=key)
    return response["Body"].read().decode("utf-8")

@tool
def query_dynamodb(table_name: str, key_name: str, key_value: str) -> dict:
    """DynamoDB 테이블에서 항목을 조회합니다.

    Args:
        table_name: 테이블 이름
        key_name: 파티션 키 이름
        key_value: 조회할 키 값

    Returns:
        조회된 항목 딕셔너리
    """
    dynamodb = boto3.resource("dynamodb")
    table = dynamodb.Table(table_name)

    response = table.get_item(Key={key_name: key_value})
    return response.get("Item", {})

@tool
def list_s3_files(bucket: str, prefix: str = "") -> list:
    """S3 버킷의 파일 목록을 가져옵니다.

    Args:
        bucket: S3 버킷 이름
        prefix: 경로 접두사 (선택)

    Returns:
        파일 키 목록
    """
    s3 = boto3.client("s3")
    response = s3.list_objects_v2(Bucket=bucket, Prefix=prefix)
    return [obj["Key"] for obj in response.get("Contents", [])]

# AWS 연동 에이전트
aws_agent = Agent(
    system_prompt="""당신은 AWS 데이터 분석 어시스턴트입니다.
S3와 DynamoDB에서 데이터를 가져와서 분석합니다.""",
    tools=[read_s3_file, query_dynamodb, list_s3_files]
)

aws_agent("my-bucket에 있는 파일 목록 보여줘")

---

5단계 — 스트리밍 응답

from strands import Agent

agent = Agent()

# 스트리밍으로 실시간 출력
for event in agent.stream_async("긴 글 써줘"):
    if "data" in event:
        print(event["data"], end="", flush=True)

---

6단계 — 멀티에이전트

에이전트가 다른 에이전트를 툴로 써요.

from strands import Agent, tool
from strands.models import AnthropicModel

model = AnthropicModel(
    client_params={"api_key": "sk-ant-xxxxx"},
    model_id="claude-sonnet-4-6",
)

# 리서치 전문 에이전트
research_agent = Agent(
    model=model,
    system_prompt="""당신은 웹 리서치 전문가입니다.
주어진 주제를 깊이 있게 조사하고 핵심 정보를 추출합니다.""",
    tools=[http_request_tool],  # 웹 요청 툴
)

# 글쓰기 전문 에이전트
writing_agent = Agent(
    model=model,
    system_prompt="""당신은 전문 기술 작가입니다.
주어진 정보를 바탕으로 명확하고 읽기 좋은 글을 작성합니다.""",
)

# 리서치 에이전트를 툴로 감싸기
@tool
def research(topic: str) -> str:
    """주제를 깊이 있게 리서치합니다.

    Args:
        topic: 조사할 주제

    Returns:
        리서치 결과 요약
    """
    return str(research_agent(f"{topic}에 대해 조사해줘"))

# 총괄 에이전트
orchestrator = Agent(
    model=model,
    system_prompt="""당신은 팀 리더입니다.
리서치 에이전트에게 정보 수집을 맡기고
그 결과를 바탕으로 최종 보고서를 작성합니다.""",
    tools=[research],
)

# 실행
orchestrator("AI 에이전트 프레임워크 트렌드 분석 보고서 작성해줘")

Swarm 패턴

from strands.multiagent import Swarm

# 같은 에이전트 여러 개로 병렬 처리
swarm = Swarm(
    agent=my_agent,
    max_agents=3  # 최대 3개 병렬
)

# 여러 작업 동시 처리
results = swarm.run([
    "Python RAG 구현 방법",
    "JavaScript RAG 구현 방법",
    "Rust RAG 구현 방법",
])

---

7단계 — MCP 서버 연결

from strands import Agent
from strands.tools.mcp import MCPClient
from mcp import StdioServerParameters

# GitHub MCP 서버 연결
github_mcp = MCPClient(
    lambda: StdioServerParameters(
        command="npx",
        args=["-y", "@modelcontextprotocol/server-github"],
        env={"GITHUB_PERSONAL_ACCESS_TOKEN": "ghp_xxxxx"},
    )
)

# MCP 툴을 에이전트에 연결
with github_mcp:
    agent = Agent(
        system_prompt="GitHub 레포 관리 어시스턴트입니다.",
        tools=github_mcp.list_tools_sync()
    )

    agent("strands-agents/sdk-python 레포의 최근 이슈 5개 알려줘")

---

8단계 — Lambda 배포

# lambda_handler.py
from strands import Agent
from strands.models.bedrock import BedrockModel
from strands_tools import calculator

# Lambda 콜드 스타트 최소화를 위해 전역에 선언
model = BedrockModel(
    model_id="us.anthropic.claude-sonnet-4-20250514",
    region_name="us-west-2",
)

agent = Agent(
    model=model,
    system_prompt="당신은 유용한 AI 어시스턴트입니다.",
    tools=[calculator],
)

def handler(event, context):
    """Lambda 핸들러"""

    # API Gateway에서 오는 경우
    user_message = event.get("body", "") or event.get("message", "")

    if not user_message:
        return {
            "statusCode": 400,
            "body": "메시지가 없습니다"
        }

    # 에이전트 실행
    response = agent(user_message)

    return {
        "statusCode": 200,
        "headers": {"Content-Type": "application/json"},
        "body": str(response)
    }

배포:

# 의존성 패키징
pip install strands-agents strands-agents-tools -t package/
cd package && zip -r ../lambda.zip . && cd ..
zip lambda.zip lambda_handler.py

# Lambda 함수 생성
aws lambda create-function \
    --function-name strands-agent \
    --runtime python3.12 \
    --handler lambda_handler.handler \
    --zip-file fileb://lambda.zip \
    --role arn:aws:iam::xxxxx:role/lambda-bedrock-role \
    --timeout 300 \
    --memory-size 512

IAM 역할에 필요한 권한:

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": [
        "bedrock:InvokeModel",
        "bedrock:InvokeModelWithResponseStream"
      ],
      "Resource": "*"
    }
  ]
}

---

전체 실전 예시 — AWS 인프라 분석 에이전트

from strands import Agent, tool
from strands.models import AnthropicModel
import boto3

model = AnthropicModel(
    client_params={"api_key": "sk-ant-xxxxx"},
    model_id="claude-sonnet-4-6",
)

@tool
def list_ec2_instances(region: str = "ap-northeast-2") -> list:
    """EC2 인스턴스 목록을 조회합니다.

    Args:
        region: AWS 리전 (기본: 서울)
    """
    ec2 = boto3.client("ec2", region_name=region)
    response = ec2.describe_instances()

    instances = []
    for reservation in response["Reservations"]:
        for instance in reservation["Instances"]:
            instances.append({
                "id": instance["InstanceId"],
                "type": instance["InstanceType"],
                "state": instance["State"]["Name"],
                "name": next(
                    (tag["Value"] for tag in instance.get("Tags", [])
                     if tag["Key"] == "Name"),
                    "unnamed"
                )
            })
    return instances

@tool
def get_cloudwatch_metrics(
    instance_id: str,
    metric_name: str = "CPUUtilization"
) -> dict:
    """CloudWatch에서 EC2 메트릭을 가져옵니다.

    Args:
        instance_id: EC2 인스턴스 ID
        metric_name: 메트릭 이름
    """
    from datetime import datetime, timedelta
    cw = boto3.client("cloudwatch")

    response = cw.get_metric_statistics(
        Namespace="AWS/EC2",
        MetricName=metric_name,
        Dimensions=[{"Name": "InstanceId", "Value": instance_id}],
        StartTime=datetime.utcnow() - timedelta(hours=1),
        EndTime=datetime.utcnow(),
        Period=300,
        Statistics=["Average"],
    )

    points = response["Datapoints"]
    if not points:
        return {"average": 0, "max": 0}

    values = [p["Average"] for p in points]
    return {
        "average": sum(values) / len(values),
        "max": max(values),
    }

# 인프라 분석 에이전트
infra_agent = Agent(
    model=model,
    system_prompt="""당신은 AWS 인프라 분석 전문가입니다.
EC2 인스턴스 상태와 메트릭을 분석해서 이상 징후를 찾고
최적화 방안을 제안합니다.""",
    tools=[list_ec2_instances, get_cloudwatch_metrics],
)

infra_agent("""
현재 실행 중인 EC2 인스턴스들을 조회하고
CPU 사용률이 높은 인스턴스가 있으면 알려줘.
최적화 방안도 제안해줘.
""")

 

 

📌 이전 글: Strands Agents 1편 — AWS가 만든 AI 에이전트 프레임워크

Strands Agents 완전 가이드 1편 — AWS가 만든 AI 에이전트 프레임워크