Claude API로 나만의 AI 에이전트 만들기 — Python SDK 실전 튜토리얼

1

pip install anthropic

1

pip install anthropic python-dotenv  # 환경변수 관리용

1
2
3
4
5

# Linux/macOS
export ANTHROPIC_API_KEY="sk-ant-..."

# Windows PowerShell
$env:ANTHROPIC_API_KEY = "sk-ant-..."

1

ANTHROPIC_API_KEY=sk-ant-...

1
2

from dotenv import load_dotenv
load_dotenv()

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

import anthropic

client = anthropic.Anthropic()

message = client.messages.create(
    model="claude-opus-4-5",
    max_tokens=1024,
    messages=[
        {"role": "user", "content": "Python에서 피보나치 수열을 생성하는 함수를 작성해줘"}
    ]
)

print(message.content[0].text)

1
2
3
4

print(message.id)           # 메시지 ID
print(message.model)        # 사용된 모델
print(message.stop_reason)  # 종료 이유 (end_turn, tool_use 등)
print(message.usage)        # 토큰 사용량

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35

import anthropic

client = anthropic.Anthropic()

def chat(messages: list, user_input: str) -> str:
    """멀티턴 대화 함수"""
    messages.append({
        "role": "user",
        "content": user_input
    })

    response = client.messages.create(
        model="claude-opus-4-5",
        max_tokens=1024,
        system="당신은 친절한 Python 코딩 어시스턴트입니다.",
        messages=messages
    )

    assistant_message = response.content[0].text
    messages.append({
        "role": "assistant",
        "content": assistant_message
    })

    return assistant_message


# 사용 예시
conversation = []

response1 = chat(conversation, "리스트 컴프리헨션이 뭐야?")
print(f"Claude: {response1}\n")

response2 = chat(conversation, "방금 설명한 내용으로 예시 코드 보여줘")
print(f"Claude: {response2}")

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35

tools = [
    {
        "name": "get_weather",
        "description": "특정 도시의 현재 날씨를 조회합니다.",
        "input_schema": {
            "type": "object",
            "properties": {
                "city": {
                    "type": "string",
                    "description": "날씨를 조회할 도시 이름 (예: Seoul, Tokyo)"
                },
                "unit": {
                    "type": "string",
                    "enum": ["celsius", "fahrenheit"],
                    "description": "온도 단위"
                }
            },
            "required": ["city"]
        }
    },
    {
        "name": "calculate",
        "description": "수학 계산을 수행합니다.",
        "input_schema": {
            "type": "object",
            "properties": {
                "expression": {
                    "type": "string",
                    "description": "계산할 수식 (예: '2 + 3 * 4')"
                }
            },
            "required": ["expression"]
        }
    }
]

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47

import math

def get_weather(city: str, unit: str = "celsius") -> dict:
    """날씨 API 호출 (예시 - 실제 API 연동 필요)"""
    # 실제 구현에서는 OpenWeatherMap 등 API 사용
    weather_data = {
        "Seoul": {"temp": 18, "condition": "맑음", "humidity": 60},
        "Tokyo": {"temp": 22, "condition": "흐림", "humidity": 75},
        "New York": {"temp": 15, "condition": "비", "humidity": 80},
    }

    data = weather_data.get(city, {"temp": 20, "condition": "알 수 없음", "humidity": 50})

    if unit == "fahrenheit":
        data["temp"] = data["temp"] * 9/5 + 32

    return {
        "city": city,
        "temperature": data["temp"],
        "unit": unit,
        "condition": data["condition"],
        "humidity": data["humidity"]
    }


def calculate(expression: str) -> dict:
    """수식 계산"""
    try:
        # 안전한 계산 (eval 대신 제한된 표현식만 허용)
        allowed_names = {k: v for k, v in math.__dict__.items() if not k.startswith("_")}
        result = eval(expression, {"__builtins__": {}}, allowed_names)
        return {"result": result, "expression": expression}
    except Exception as e:
        return {"error": str(e), "expression": expression}


def execute_tool(tool_name: str, tool_input: dict) -> str:
    """도구 실행 디스패처"""
    if tool_name == "get_weather":
        result = get_weather(**tool_input)
    elif tool_name == "calculate":
        result = calculate(**tool_input)
    else:
        result = {"error": f"알 수 없는 도구: {tool_name}"}

    import json
    return json.dumps(result, ensure_ascii=False)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86

import anthropic
import json

client = anthropic.Anthropic()


def run_agent(user_request: str, tools: list) -> str:
    """
    에이전트 메인 루프

    Args:
        user_request: 사용자 요청
        tools: 사용 가능한 도구 목록

    Returns:
        최종 응답 텍스트
    """
    messages = [
        {"role": "user", "content": user_request}
    ]

    print(f"사용자: {user_request}\n")

    while True:
        # Claude API 호출
        response = client.messages.create(
            model="claude-opus-4-5",
            max_tokens=4096,
            tools=tools,
            messages=messages
        )

        print(f"[stop_reason: {response.stop_reason}]")

        # 최종 응답인 경우 루프 종료
        if response.stop_reason == "end_turn":
            # 텍스트 응답 추출
            final_response = ""
            for block in response.content:
                if hasattr(block, "text"):
                    final_response += block.text
            return final_response

        # 도구 사용 요청인 경우
        if response.stop_reason == "tool_use":
            # 어시스턴트 응답을 메시지 히스토리에 추가
            messages.append({
                "role": "assistant",
                "content": response.content
            })

            # 각 도구 실행
            tool_results = []
            for block in response.content:
                if block.type == "tool_use":
                    print(f"도구 실행: {block.name}({block.input})")

                    result = execute_tool(block.name, block.input)
                    print(f"도구 결과: {result}\n")

                    tool_results.append({
                        "type": "tool_result",
                        "tool_use_id": block.id,
                        "content": result
                    })

            # 도구 결과를 메시지에 추가
            messages.append({
                "role": "user",
                "content": tool_results
            })
        else:
            # 예상치 못한 stop_reason
            print(f"예상치 못한 stop_reason: {response.stop_reason}")
            break

    return "에이전트 실행 중 오류가 발생했습니다."


# 실행 예시
if __name__ == "__main__":
    result = run_agent(
        "서울과 도쿄의 날씨를 비교해주고, 두 도시 온도 차이를 계산해줘",
        tools
    )
    print(f"\n최종 답변:\n{result}")

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

사용자: 서울과 도쿄의 날씨를 비교해주고, 두 도시 온도 차이를 계산해줘

[stop_reason: tool_use]
도구 실행: get_weather({'city': 'Seoul', 'unit': 'celsius'})
도구 결과: {"city": "Seoul", "temperature": 18, ...}

도구 실행: get_weather({'city': 'Tokyo', 'unit': 'celsius'})
도구 결과: {"city": "Tokyo", "temperature": 22, ...}

도구 실행: calculate({'expression': '22 - 18'})
도구 결과: {"result": 4, "expression": "22 - 18"}

[stop_reason: end_turn]

최종 답변:
서울은 현재 18°C 맑음, 도쿄는 22°C 흐림입니다. 두 도시의 온도 차이는 4°C로...

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

def stream_response(user_input: str) -> None:
    """스트리밍 응답 출력"""
    with client.messages.stream(
        model="claude-opus-4-5",
        max_tokens=1024,
        messages=[{"role": "user", "content": user_input}]
    ) as stream:
        for text in stream.text_stream:
            print(text, end="", flush=True)
    print()  # 줄바꿈


stream_response("Python의 제너레이터를 간단하게 설명해줘")

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41

def create_coding_agent(language: str = "Python") -> callable:
    """특정 언어 전문 코딩 에이전트 생성"""
    system_prompt = f"""당신은 {language} 전문 시니어 엔지니어입니다.

역할:
- 코드 리뷰 및 개선 제안
- 버그 분석 및 수정 방법 설명
- 베스트 프랙티스 안내
- 성능 최적화 조언

원칙:
- 항상 코드 예시와 함께 설명합니다
- 이유를 설명하지 않고 답변하지 않습니다
- 보안과 성능 양쪽을 고려합니다
- 간결하고 읽기 쉬운 코드를 지향합니다"""

    def agent(user_input: str, messages: list = None) -> str:
        if messages is None:
            messages = []

        messages.append({"role": "user", "content": user_input})

        response = client.messages.create(
            model="claude-opus-4-5",
            max_tokens=2048,
            system=system_prompt,
            messages=messages
        )

        reply = response.content[0].text
        messages.append({"role": "assistant", "content": reply})
        return reply

    return agent


# 사용 예시
python_expert = create_coding_agent("Python")
conversation = []

print(python_expert("이 코드의 문제점이 뭐야?\n\ndef get_users():\n    users = db.execute('SELECT * FROM users')\n    return users", conversation))

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50

import os

file_tools = [
    {
        "name": "read_file",
        "description": "파일 내용을 읽습니다.",
        "input_schema": {
            "type": "object",
            "properties": {
                "path": {"type": "string", "description": "파일 경로"}
            },
            "required": ["path"]
        }
    },
    {
        "name": "write_file",
        "description": "파일에 내용을 씁니다.",
        "input_schema": {
            "type": "object",
            "properties": {
                "path": {"type": "string"},
                "content": {"type": "string"}
            },
            "required": ["path", "content"]
        }
    },
    {
        "name": "list_files",
        "description": "디렉터리의 파일 목록을 반환합니다.",
        "input_schema": {
            "type": "object",
            "properties": {
                "directory": {"type": "string"}
            },
            "required": ["directory"]
        }
    }
]

def execute_file_tool(tool_name: str, tool_input: dict) -> str:
    if tool_name == "read_file":
        with open(tool_input["path"], "r", encoding="utf-8") as f:
            return f.read()
    elif tool_name == "write_file":
        with open(tool_input["path"], "w", encoding="utf-8") as f:
            f.write(tool_input["content"])
        return f"파일 작성 완료: {tool_input['path']}"
    elif tool_name == "list_files":
        files = os.listdir(tool_input["directory"])
        return "\n".join(files)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

import time
from anthropic import RateLimitError, APIError

def robust_agent_call(messages: list, tools: list, max_retries: int = 3) -> any:
    """재시도 로직이 포함된 API 호출"""
    for attempt in range(max_retries):
        try:
            return client.messages.create(
                model="claude-opus-4-5",
                max_tokens=4096,
                tools=tools,
                messages=messages
            )
        except RateLimitError:
            if attempt < max_retries - 1:
                wait_time = 2 ** attempt  # 지수 백오프
                print(f"Rate limit 도달. {wait_time}초 후 재시도...")
                time.sleep(wait_time)
            else:
                raise
        except APIError as e:
            print(f"API 오류: {e}")
            raise

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

class TokenTracker:
    def __init__(self):
        self.input_tokens = 0
        self.output_tokens = 0

    def add(self, response):
        self.input_tokens += response.usage.input_tokens
        self.output_tokens += response.usage.output_tokens

    def estimate_cost(self, model: str = "claude-opus-4-5") -> float:
        """대략적인 비용 계산 (달러)"""
        # Claude Opus 4.5 기준 (2026년 가격 참고)
        input_cost = self.input_tokens * 15 / 1_000_000
        output_cost = self.output_tokens * 75 / 1_000_000
        return input_cost + output_cost

    def report(self):
        print(f"입력 토큰: {self.input_tokens:,}")
        print(f"출력 토큰: {self.output_tokens:,}")
        print(f"예상 비용: ${self.estimate_cost():.4f}")


tracker = TokenTracker()
response = client.messages.create(...)
tracker.add(response)
tracker.report()