목차
- 문제와 접근
- HTTP 기초 개념
- requests 라이브러리 기본 사용
- Session 객체
- rank_users_by_posts — 실습 함수
- 페이지 기반 페이지네이션 (Page-based)
- 커서 기반 페이지네이션 (Cursor-based)
- LLM 도구로 감싸기 (@tool + requests)
- 오케스트레이터 패턴
- build_user_profile / build_team_report
- 핵심 개념 총정리
- 자주 나오는 실수/주의사항
- 외부 API의 변화를 흡수할 경계
1. 문제와 접근
HTTP API를 LLM 도구로 감싸기 전에 네트워크 호출 자체의 성공 조건부터 정해야 한다. 상태 코드, 응답 파싱, 페이지 순회를 단계별로 확인하고 애플리케이션이 사용할 데이터 형태로 변환한다. 모델에는 이 경계가 정리된 도구를 제공하는 흐름으로 접근한다.
실습 API
JSONPlaceholder — 무료 테스트용 REST API
BASE_URL = "https://jsonplaceholder.typicode.com"
| 엔드포인트 | 설명 |
|---|
| /posts/{id} | 게시글 조회 |
| /users/{id} | 사용자 조회 |
| /todos | 할일 목록 |
| /posts?userId=1 | 특정 사용자의 게시글 |
2. HTTP 기초 개념
요청(Request) 구성 요소
1
2
3
4
| Method : 뭘 할건지? GET / POST / PUT / DELETE
URL : 어디서? https://jsonplaceholder.typicode.com/posts
헤더 : 니가 누군데? {'Authorization': 'Bearer ...', 'User-Agent': 'chrome/...'}
바디 : 뭘 보낼건지? {'title': '제목입니다', 'content': '내용입니다'}
|
상태 코드
1
2
3
4
5
6
7
8
9
10
| 성공
200 : 조회 성공 (OK)
201 : 생성 성공 (Created)
실패
400 : 요청 형식 틀림 (Bad Request)
401 : 인증 없음 (Unauthorized)
403 : 권한 없음 (Forbidden)
404 : 리소스 없음 (Not Found)
500 : 서버 내부 오류 (Internal Server Error)
|
3. requests 라이브러리 기본 사용
요청을 보낸 뒤에는 네트워크 실패, 실패 상태 코드, 예상과 다른 응답 본문을 구분한다. timeout과 응답 검증을 호출 경계에 두면 상위 로직이 외부 서비스의 예외 형식에 직접 의존하는 범위를 줄일 수 있다.
1
2
3
| import requests
BASE_URL = "https://jsonplaceholder.typicode.com"
|
GET — 데이터 조회
1
2
3
4
5
| r = requests.get(f"{BASE_URL}/posts/1")
r.status_code # → 200
r.json() # → {'userId': 1, 'id': 1, 'title': '...', 'body': '...'}
r.ok # → True (2xx이면 True)
|
POST — 데이터 생성
1
2
3
4
5
6
7
| r = requests.post(
f"{BASE_URL}/posts",
json={'title': '새 글', 'body': '본문', 'userId': 1}
)
r.status_code # → 201
r.json() # → {'title': '새 글', 'body': '본문', 'userId': 1, 'id': 101}
|
DELETE — 데이터 삭제
1
2
3
4
| r = requests.delete(f"{BASE_URL}/posts/1", timeout=5)
r.status_code # → 200
r.json() # → {}
|
4. Session 객체
Session을 사용하면 헤더를 한 번만 설정하고 여러 요청에 재사용할 수 있다.
1
2
3
4
5
6
7
8
9
10
11
12
13
| session = requests.Session()
session.headers.update({
'User-Agent': 'myapp/1.0',
'Accept' : 'application/json'
})
# 이후 모든 요청에 헤더 자동 포함
r = session.get(f"{BASE_URL}/posts", params={"userId": 1})
len(r.json()) # → 10
r2 = session.get(f"{BASE_URL}/users/1")
r2.json()['name'], r2.json()['email']
# → ('Leanne Graham', 'Sincere@april.biz')
|
params — 쿼리 파라미터
1
2
3
| # ?userId=1 을 params 딕셔너리로 전달
r = session.get(f"{BASE_URL}/posts", params={"userId": 1})
# 실제 URL: https://jsonplaceholder.typicode.com/posts?userId=1
|
5. rank_users_by_posts — 실습 함수
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
| def rank_users_by_posts(user_ids):
s = requests.Session()
s.headers.update({'User-Agent': 'myapp/1.0', 'Accept': 'application/json'})
counts = []
for uid in user_ids:
r = s.get(f"{BASE_URL}/posts", params={"userId": uid})
counts.append({"userId": uid, "post_count": len(r.json())})
counts.sort(key=lambda x: x["post_count"], reverse=True)
for i, item in enumerate(counts, 1):
item['rank'] = i
return counts
rank_users_by_posts([1, 2, 3, 4, 5])
# → [{'userId': 1, 'post_count': 10, 'rank': 1}, ...]
|
6. 페이지 기반 페이지네이션 (Page-based)
데이터가 많을 때 여러 페이지로 나눠서 가져온다.
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
| # ?_page=1&_limit=10 형태로 페이지 지정
def fetch_all_posts(page_size=10, max_pages=100):
page = 1
while page <= max_pages:
r = requests.get(
f"{BASE_URL}/posts",
params={"_page": page, "_limit": page_size}
)
items = r.json()
if not items: # 빈 응답 → 마지막 페이지
break
print(f" page {page}: {len(items)}개")
for item in items:
yield item # generator로 하나씩 반환
if len(items) < page_size: # 데이터가 page_size보다 적으면 마지막 페이지
break
page += 1
all_posts = list(fetch_all_posts(page_size=20))
# page 1: 20개 page 2: 20개 ... page 5: 20개
len(all_posts) # → 100
|
필터링 + 페이지네이션 결합
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
| def collect_completed_todos(max_items=20, page_size=10):
collected = []
page = 1
while len(collected) < max_items:
r = requests.get(f"{BASE_URL}/todos",
params={"_page": page, "_limit": page_size})
items = r.json()
if not items:
break
completed = [t for t in items if t.get('completed')]
collected.extend(completed[:max_items - len(collected)])
print(f" page {page}: {len(items)}개, 수집: {len(collected)}개")
if len(items) < page_size:
break
page += 1
return collected[:max_items]
|
7. 커서 기반 페이지네이션 (Cursor-based)
페이지 순회에서는 다음 커서가 없을 때만 아니라 같은 커서가 반복되는 상황도 확인한다. 수집 중 데이터가 바뀔 수 있는 API라면 중복 ID와 누락 가능성을 따로 점검하고, 종료 조건 없이 반복하지 않도록 한도를 둔다.
page 번호 대신 “마지막으로 본 항목의 위치”를 커서로 사용한다.
1
2
3
4
5
6
7
8
9
10
11
12
13
| def cursor_paginate_simulator(total_items=25, page_size=10):
all_data = [{"id": i, "name": f"item-{i}"} for i in range(1, total_items+1)]
def fetch_page(cursor=None, limit=page_size):
start = int(cursor) if cursor else 0
end = min(start + limit, total_items)
items = all_data[start:end]
next_cursor = str(end) if end < total_items else None
return {'items': items, 'next_cursor': next_cursor}
return fetch_page
fetch = cursor_paginate_simulator(total_items=25, page_size=10)
|
커서 기반 전체 수집 루프
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
| cursor = None
page = 1
all_items = []
while True:
result = fetch(cursor=cursor)
print(f"page {page}: {len(result['items'])}개, next={result['next_cursor']}")
all_items.extend(result['items'])
if not result['next_cursor']: # None이면 마지막 페이지
break
cursor = result['next_cursor']
page += 1
# page 1: 10개, next=10
# page 2: 10개, next=20
# page 3: 5개, next=None
|
페이지 기반 vs 커서 기반 비교
1
2
3
4
5
6
7
8
9
| [페이지 기반]
- ?_page=1&_limit=10 형태
- 구현 간단, 특정 페이지로 바로 이동 가능
- 데이터 변동 시 중복/누락 발생 가능
[커서 기반]
- 마지막 위치를 next_cursor로 추적
- 대용량 데이터에서 안정적
- 특정 페이지로 바로 이동 불가
|
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
| @tool
def fetch_user(user_id):
"""지정된 ID의 사용자 정보를 조회합니다"""
r = requests.get(f"{BASE_URL}/users/{user_id}")
if not r.ok:
return f"error: status {r.status_code}"
data = r.json()
return json.dumps({
'name' : data['name'],
'email' : data['email'],
'phone' : data['phone'],
'company': data['company']['name']
}, ensure_ascii=False)
@tool
def fetch_user_post(user_id):
"""지정된 ID의 사용자가 작성한 게시글을 조회합니다"""
r = requests.get(f"{BASE_URL}/posts", params={'userId': user_id})
if not r.ok:
return f"error: status {r.status_code}"
posts = r.json()
return json.dumps(
[{'id': p['id'], 'title': p['title']} for p in posts],
ensure_ascii=False
)
|
실행 확인
1
2
3
4
5
6
7
8
| llm_api = llm.bind_tools([fetch_user, fetch_user_post])
response = llm_api.invoke("1번 유저 정보와 그 사람이 쓴 게시글 갯수 알려줘")
print(f"호출된 도구 개수: {len(response.tool_calls)}") # → 2
for tc in response.tool_calls:
print(f"{tc['name']}: {tc['args']}")
# fetch_user: {'user_id': 1}
# fetch_user_post: {'user_id': 1}
|
9. 오케스트레이터 패턴
도구 실행 + 에러 처리 + 로그를 통합한 패턴
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
| def orchestrate(query, max_turns=5):
llm_api = llm.bind_tools(api_tools)
messages = [HumanMessage(content=query)]
tool_call_count = 0
MAX_TOOL_CALLS = 10
for turn in range(max_turns):
try:
resp = llm_api.invoke(messages)
except Exception as e:
return f"llm call fail: {e}"
messages.append(resp)
if not resp.tool_calls:
return resp.content
for tc in resp.tool_calls:
name = tc['name']
args = tc['args']
if name not in tool_map:
result = f"Error: unknown tool '{name}'"
else:
tool_call_count += 1
if tool_call_count > MAX_TOOL_CALLS:
return "Error: tool call limit exceeded"
try:
result = tool_map[tc['name']].invoke(tc['args'])
except:
result = "Error"
print(f" [Turn {turn+1}] {tc['name']} ({tc['args']}) → {result[:80]}")
messages.append(ToolMessage(content=result, tool_call_id=tc['id']))
return '최대 턴 초과'
|
10. build_user_profile / build_team_report
사용자 프로필 빌더
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
| def build_user_profile(user_id):
profile = {'user_id': user_id}
try: # 기본 정보
r = requests.get(f"{BASE_URL}/users/{user_id}")
if r.ok:
u = r.json()
profile['name'] = u['name']
profile['email'] = u['email']
profile['city'] = u['address']['city']
except:
profile['user_error'] = 'error'
try: # 게시글
r = requests.get(f"{BASE_URL}/posts", params={'userId': user_id})
if r.ok:
posts = r.json()
profile['post_count'] = len(posts)
profile['recent_posts'] = [p['title'][:40] for p in posts[:3]]
except Exception as e:
profile['post_error'] = str(e)
try: # 할일
r = requests.get(f"{BASE_URL}/todos", params={'userId': user_id})
if r.ok:
todos = r.json()
profile['todo_total'] = len(todos)
profile['todo_completed'] = sum(1 for t in todos if t['completed'])
except:
profile['todo_error'] = 'error'
return profile
|
중첩 딕셔너리 경로 접근
1
2
3
4
5
6
7
8
9
10
11
12
| def get_by_path(data, path, default=None):
current = data
for key in path.split("."): # "address.city" → ["address", "city"]
if isinstance(current, dict) and key in current:
current = current[key]
else:
return default
return current
user_data = requests.get(f"{BASE_URL}/users/1").json()
get_by_path(user_data, 'name') # → 'Leanne Graham'
get_by_path(user_data, 'address.city') # → 'Gwenborough'
|
11. 핵심 개념 총정리
HTTP 메서드 정리
| 메서드 | 용도 | 성공 상태코드 |
|---|
| GET | 데이터 조회 | 200 |
| POST | 데이터 생성 | 201 |
| PUT | 데이터 수정 | 200 |
| DELETE | 데이터 삭제 | 200 |
requests 주요 메서드 정리
| 메서드 | 설명 |
|---|
| requests.get(url, params=…) | GET 요청 |
| requests.post(url, json=…) | POST 요청 (json 바디) |
| requests.delete(url, timeout=…) | DELETE 요청 |
| r.status_code | 상태 코드 |
| r.json() | JSON 파싱 결과 |
| r.ok | 2xx이면 True |
| requests.Session() | 세션 생성 (헤더 재사용) |
12. 자주 나오는 실수/주의사항
실수 1: params 대신 URL에 직접 작성
1
2
3
4
5
| # 번거롭고 오타 위험
r = requests.get(f"{BASE_URL}/posts?userId=1&_page=1&_limit=10")
# 올바른 코드 — params 딕셔너리 사용
r = requests.get(f"{BASE_URL}/posts", params={"userId": 1, "_page": 1, "_limit": 10})
|
실수 2: r.ok 확인 없이 r.json() 바로 사용
1
2
3
4
5
6
7
8
| # 틀린 코드 → 에러 응답에서 json() 실패 가능
data = requests.get(f"{BASE_URL}/users/9999").json()
# 올바른 코드
r = requests.get(f"{BASE_URL}/users/9999")
if not r.ok:
return f"error: status {r.status_code}"
data = r.json()
|
실수 3: 페이지네이션에서 마지막 페이지 조건 빠뜨리기
1
2
3
4
5
6
7
8
9
| # 틀린 코드 → 빈 응답 체크만 하면 무한 루프 가능
while True:
items = r.json()
if not items: break
# page_size보다 적은 경우 처리 없음 → 다음 페이지 요청 계속
# 올바른 코드
if not items or len(items) < page_size:
break
|
실수 4: timeout 미설정
1
2
3
4
5
| # 틀린 코드 → 서버 응답 없으면 무한 대기
r = requests.get(url)
# 올바른 코드
r = requests.get(url, timeout=5) # 5초 후 TimeoutError
|
13. 외부 API의 변화를 흡수할 경계
외부 응답을 그대로 도구 반환값으로 쓰기보다 필요한 필드와 오류 상태를 일정한 형태로 정리한다. 이렇게 하면 API의 컬럼 추가나 오류 메시지 변경이 프롬프트까지 바로 전달되는 일을 줄일 수 있다.
프로필과 팀 리포트를 만들 때는 일부 API만 성공한 경우의 결과도 정의한다. 전체 실패와 부분 결과를 구분하고, 누락된 정보를 0이나 빈 목록으로 처리해도 되는지 항목별로 판단한다.