진행 상황
기계 번역
이 페이지는 영어 문서를 자동으로 번역한 것이며, 영어 페이지가 기준이 되는 정식 버전입니다. 어색하거나 잘못된 부분이 있다면 번역 페이지에서 제보하는 방법을 확인하세요.
30초가 걸리면서 그 30초 동안 아무 말도 하지 않는 도구는 고장 난 것처럼 보입니다.
진행 상황 알림이 이 문제를 해결합니다. 도구는 얼마나 진행되었는지 보고하고, 클라이언트는 그 정보로 무엇을 그릴지 결정합니다. 진행 막대일 수도, 스피너일 수도, 로그 한 줄일 수도 있습니다.
도구에서 보고하기
Context 매개변수를 받고 report_progress를 호출하세요.
from mcp.server import MCPServer
from mcp.server.mcpserver import Context
mcp = MCPServer("Bookshop")
@mcp.tool()
async def import_catalog(urls: list[str], ctx: Context) -> str:
"""Import book records from a list of catalog URLs."""
for done, url in enumerate(urls, start=1):
await ctx.report_progress(done, total=len(urls), message=f"Imported {url}")
return f"Imported {len(urls)} records."
인자는 세 개이며, 각 인자의 의미는 직접 정합니다.
progress: 얼마나 진행했는지입니다. 사양은 보고할 때마다 이 값이 증가해야 한다고 요구합니다. 같은 값을 반복하거나 뒤로 돌아가면 안 됩니다.total: 알고 있다면, 전체가 얼마나 되는지입니다. 선택 사항입니다.message: 이 단계를 설명하는, 사람이 읽을 수 있는 한 줄입니다. 선택 사항입니다.
ctx는 타입 힌트 덕분에 주입되며 모델에는 전혀 보이지 않습니다. import_catalog의 입력 스키마에는 urls 속성 하나만 있습니다. Context 페이지는 이 객체를 본격적으로 다루며, 진행 상황 보고는 이 객체가 제공하는 기능 중 하나입니다.
클라이언트에서 수신하기
클라이언트는 호출 단위로 수신을 선택합니다. call_tool에 progress_callback= 인자를 전달하면 됩니다.
import anyio
from mcp import Client
from server import mcp
async def show(progress: float, total: float | None, message: str | None) -> None:
print(f"{message} ({progress}/{total})")
async def main() -> None:
async with Client(mcp) as client:
result = await client.call_tool(
"import_catalog",
{"urls": ["https://example.com/a.json", "https://example.com/b.json"]},
progress_callback=show,
)
print(result.structured_content)
anyio.run(main)
콜백은 서버가 보고한 값 그대로, 즉 progress, total, message를 받는 async 함수입니다.
Info
Client(mcp)는 서버 객체에 메모리 안에서 직접 연결하며, 테스트
페이지의 기반이 되는 것과 같은 클라이언트입니다. progress_callback은 Client가 어떤 트랜스포트를
쓰든 같은 매개변수입니다. 다만 곧 보게 될 타이밍은 인메모리 연결의 타이밍입니다. 인메모리 연결은
콜백을 인라인으로 실행하므로 모든 보고가 call_tool이 반환되기 전에 도착합니다. 실제 트랜스포트에서는
알림과 결과가 경쟁하므로, 느린 콜백은 call_tool이 반환된 뒤에도 여전히 실행 중일 수 있습니다.
직접 해 보기
client.py를 server.py 옆에 두고 실행하세요.
python client.py
Imported https://example.com/a.json (1/2)
Imported https://example.com/b.json (2/2)
{'result': 'Imported 2 records.'}
서버의 await ctx.report_progress(...) 하나하나가 클라이언트에서 show 호출 하나가 되었고, 순서도 그대로이며, 두 줄 모두 call_tool이 반환되기 전에 출력되었습니다. 진행 상황은 결과에 묶여 오지 않고, 도구가 아직 작업하는 동안 스트리밍됩니다.
Warning
progress_callback은 Client가 아니라 호출에 속합니다. 이를 위한 생성자 인자는 없습니다.
호출마다 원하는 콜백이 다르기 때문입니다. 어떤 호출은 다운로드 막대를 움직이고, 다음 호출은
로그 한 줄을 남깁니다.
Check
이제 progress_callback=show 부분을 지우고 다시 실행하세요.
{'result': 'Imported 2 records.'}
오류도 경고도 없고 결과는 같습니다. report_progress는 호출자가 진행 상황을 요청하지 않았으면
아무 일도 하지 않으므로, 조건 없이 보고하면 되고 누가 듣고 있는지 신경 쓸 필요가 없습니다.
전체 양을 모를 때
total은 분모를 알 때 쓰는 값입니다. 모르는 경우도 많습니다. 피드를 비우거나, 커서를 따라가거나, 길이 헤더가 없는 무언가를 내려받을 때가 그렇습니다.
그럴 때는 생략하세요.
from collections.abc import AsyncIterator
from mcp.server import MCPServer
from mcp.server.mcpserver import Context
mcp = MCPServer("Bookshop")
async def fetch_records(feed_url: str) -> AsyncIterator[str]:
for title in ("Dune", "Neuromancer", "Hyperion"):
yield f"{feed_url}#{title}"
@mcp.tool()
async def import_feed(feed_url: str, ctx: Context) -> str:
"""Import every record a catalog feed yields."""
imported = 0
async for record in fetch_records(feed_url):
imported += 1
await ctx.report_progress(imported, message=f"Imported {record}")
return f"Imported {imported} records."
콜백은 total=None 값을 받습니다. 클라이언트는 여전히 활동("3 imported so far...")은 보여 줄 수 있지만 백분율은 보여 줄 수 없습니다. 더 보기 좋은 막대를 위해 전체 양을 지어내지 마세요.
Tip
progress가 꼭 특정한 무언가를 세어야 하는 것은 아닙니다. 바이트, 행, 페이지 중 사용자가
알아볼 단위를 고르고, 지킬 수 있는 total만 약속하세요.
요약
Context를 받는 도구라면 어디서든await ctx.report_progress(progress, total=None, message=None)형태로 호출합니다.- 클라이언트는
call_tool에progress_callback=인자를 전달합니다. 호출마다 지정하며,Client에는 지정하지 않습니다. - 콜백은
async (progress, total, message) -> None형태이며 도구가 아직 실행 중인 동안 호출됩니다. - 호출에 콜백이 없으면
report_progress는 아무 일도 하지 않습니다. 조건 없이 보고하세요. total을 모르면 생략하세요. 콜백은None을 받습니다.
진행 상황은 실행 중인 도구가 사용자에게 보여 주는 것입니다. 서버를 운영하는 운영자를 위해 도구가 남기는 로그 줄은 별개의 채널이며, 로깅에서 다룹니다.