AI 에이전트가 실험도구가 되다 — MCP로 바이오 DB 검색하기

이 토픽을 마치면

교과서에서 배운 MCP(Model Context Protocol) 와 harness 를 엮어서, Claude나 GPT 같은 LLM이 자연어 지시로 NCBI Entrez, PubMed, UniProt 같은 바이오 DB를 자동 조회하도록 만드는 도구를 직접 짤 수 있습니다. LLM tool use의 내부 원리와 그 신뢰성을 평가하는 harness의 실체를 이해하게 됩니다.

이 글은 교육용 일반 예제 입니다. 실전 배포 MCP 서버는 인증, rate limiting, 관찰성이 훨씬 정교합니다.

"왜 매번 같은 검색을 다시 하지?" — 반복 실험의 함정

여러분이 논문 아이디어를 준비할 때 매일 이런 검색을 반복한다고 합시다.

NCBI PubMed에 "BRCA1" AND "review" AND "2024" 검색
초록을 훑어보며 관련성 판정
인용 관계 확인
UniProt에서 관련 단백질 도메인 확인
GenBank에서 서열 다운로드

이 조합을 매일 반복합니다. 문제가 있습니다.

문제 1: 각 DB의 검색 문법이 다릅니다. NCBI Entrez의 필터 문법, UniProt의 검색 문법, GenBank의 accession 형식이 모두 다릅니다. 매번 다시 기억해야 합니다.

문제 2: 여러 DB의 결과를 손으로 조합해야 합니다. PubMed에서 얻은 저자 이름을 UniProt에 다시 입력하고, 그 결과 accession을 GenBank에 다시 입력하는 식.

문제 3: 이 작업의 대부분이 결정적 규칙 기반 이지만 매일 다른 키워드로 반복됩니다. 즉 자동화하기에 딱 좋은 형태이지만, 각 DB의 API 문서를 익히는 오버헤드가 큽니다.

진짜 접근은 이것을 AI 에이전트에게 위임하는 것입니다. LLM에게 자연어로 요청을 주면, 뒤에서 여러 API를 순차 호출해 결과를 종합해줍니다. 이것을 안전하고 표준화된 방식으로 하는 프로토콜이 MCP(Model Context Protocol) 입니다.

블랙박스에서 부품으로 — MCP 열어보기

MCP는 표면적으로 복잡해 보이지만, 핵심 개념은 세 가지입니다.

부품 1: 도구 정의 (Tool Definition)

MCP 서버는 자기가 제공할 수 있는 도구 목록을 LLM에 제시합니다. 각 도구는 이름, 설명, 인자 스키마로 정의됩니다.

python

tool_definition = {
    "name": "search_pubmed",
    "description": "PubMed 논문 데이터베이스에서 키워드로 검색하고 상위 N개 결과를 반환",
    "input_schema": {
        "type": "object",
        "properties": {
            "query": {"type": "string", "description": "검색 쿼리, PubMed 문법 사용"},
            "max_results": {"type": "integer", "default": 10}
        },
        "required": ["query"]
    }
}

핵심은 description 이 LLM이 실제로 읽고 판단하는 유일한 정보라는 점입니다. 좋은 도구 설명이 좋은 에이전트 동작을 결정합니다.

부품 2: 실행 함수

각 도구는 실제로 실행되는 함수를 갖습니다. LLM이 도구 호출을 요청하면 이 함수가 실행되고 결과가 LLM에게 돌아갑니다.

python

async def search_pubmed(query: str, max_results: int = 10) -> dict:
    import httpx
    base = "https://eutils.ncbi.nlm.nih.gov/entrez/eutils"
    async with httpx.AsyncClient() as client:
        search_response = await client.get(
            f"{base}/esearch.fcgi",
            params={"db": "pubmed", "term": query, "retmax": max_results, "retmode": "json"}
        )
        ids = search_response.json()["esearchresult"]["idlist"]
        
        if not ids:
            return {"results": []}
        
        summary_response = await client.get(
            f"{base}/esummary.fcgi",
            params={"db": "pubmed", "id": ",".join(ids), "retmode": "json"}
        )
        summaries = summary_response.json()["result"]
        
        return {
            "results": [
                {
                    "pmid": pmid,
                    "title": summaries[pmid].get("title"),
                    "journal": summaries[pmid].get("fulljournalname"),
                    "pubdate": summaries[pmid].get("pubdate")
                }
                for pmid in ids
            ]
        }

주의점: 이 함수는 외부 API를 호출하므로 네트워크·rate limit·타임아웃을 처리해야 합니다. 아래에서 이 부분을 다룹니다.

부품 3: MCP 서버 골격

이 모두를 하나의 서버로 묶습니다. Anthropic이 공식 파이썬 SDK를 제공하지만, 개념을 명확히 하기 위해 직접 만들어 봅시다.

python

from typing import Any, Callable, Coroutine
import json

class MCPServer:
    def __init__(self):
        self.tools: dict[str, dict[str, Any]] = {}
        self.handlers: dict[str, Callable[..., Coroutine]] = {}
    
    def register(self, definition: dict, handler: Callable[..., Coroutine]):
        name = definition["name"]
        self.tools[name] = definition
        self.handlers[name] = handler
    
    def list_tools(self) -> list[dict]:
        return list(self.tools.values())
    
    async def call_tool(self, name: str, arguments: dict) -> dict:
        if name not in self.handlers:
            return {"error": f"Unknown tool: {name}"}
        try:
            return await self.handlers[name](**arguments)
        except Exception as e:
            return {"error": str(e)}


server = MCPServer()
server.register(
    definition={
        "name": "search_pubmed",
        "description": "Search PubMed for scientific papers",
        "input_schema": {
            "type": "object",
            "properties": {
                "query": {"type": "string"},
                "max_results": {"type": "integer", "default": 10}
            },
            "required": ["query"]
        }
    },
    handler=search_pubmed
)

이 서버가 LLM 클라이언트와 통신하려면 표준 입출력 또는 HTTP를 통해야 하지만, 여기서는 함수로 직접 호출 하는 형태로 개념을 이해합니다.

LLM 통합 — Claude와 함께 쓰기

이제 이 서버를 LLM 호출과 연결합니다. Claude API의 tool use 기능을 씁니다.

python

import anthropic

client = anthropic.Anthropic()

async def agent_loop(user_message: str, server: MCPServer, max_iterations: int = 5) -> str:
    messages = [{"role": "user", "content": user_message}]
    tools = [
        {
            "name": t["name"],
            "description": t["description"],
            "input_schema": t["input_schema"]
        }
        for t in server.list_tools()
    ]
    
    for iteration in range(max_iterations):
        response = client.messages.create(
            model="claude-opus-4-8",
            max_tokens=2048,
            tools=tools,
            messages=messages
        )
        
        if response.stop_reason == "end_turn":
            text_blocks = [b.text for b in response.content if b.type == "text"]
            return "\n".join(text_blocks)
        
        if response.stop_reason == "tool_use":
            tool_uses = [b for b in response.content if b.type == "tool_use"]
            messages.append({"role": "assistant", "content": response.content})
            
            tool_results = []
            for tool_use in tool_uses:
                result = await server.call_tool(tool_use.name, tool_use.input)
                tool_results.append({
                    "type": "tool_result",
                    "tool_use_id": tool_use.id,
                    "content": json.dumps(result)
                })
            
            messages.append({"role": "user", "content": tool_results})
    
    return "Max iterations reached"

이제 사용:

python

result = await agent_loop(
    "BRCA1 유전자의 2024년 이후 리뷰 논문 3편을 찾아서 제목과 저널만 알려줘.",
    server
)
print(result)

LLM이 자동으로 search_pubmed(query="BRCA1 AND review AND 2024:2025", max_results=3) 을 호출하고, 결과를 자연어로 정리해 반환합니다.

페이딩 — 여러분이 채워야 할 세 빈칸

빈칸 1: 도구 확장 — UniProt 검색

PubMed만 있는 서버는 절반짜리입니다. UniProt 도구를 추가합시다.

python

async def search_uniprot(query: str, max_results: int = 10) -> dict:
    import httpx
    async with httpx.AsyncClient() as client:
        # TODO: UniProt REST API 호출
        # 엔드포인트: https://rest.uniprot.org/uniprotkb/search
        # 파라미터: query, format=json, size=max_results
        # 결과: {"results": [{"accession": ..., "name": ..., "gene": ...}]}
        pass

힌트: UniProt 응답의 results 배열 각 항목에서 primaryAccession, proteinDescription.recommendedName.fullName.value, genes[0].geneName.value 를 추출.

빈칸 2: harness — 에이전트 평가 시스템

에이전트가 잘 작동하는지 검증하려면 evaluation harness가 필요합니다. 여러 테스트 케이스를 자동으로 돌려 각 결과를 평가합니다.

python

async def evaluate_agent(
    test_cases: list[dict],
    server: MCPServer
) -> dict:
    """
    test_cases: [
        {
            "prompt": "사용자 요청",
            "expected_tool_calls": ["search_pubmed"],
            "expected_content_contains": ["BRCA1", "review"]
        }
    ]
    """
    results = []
    for tc in test_cases:
        # TODO: agent_loop 실행, 실제 호출된 도구 추적, 결과 검증
        # 통과 여부를 results에 기록
        pass
    
    return {
        "total": len(test_cases),
        "passed": sum(1 for r in results if r["passed"]),
        "details": results
    }

힌트: agent_loop 내부에서 도구 호출을 로그로 기록하도록 수정. 각 호출을 리스트에 append하고 반환.

빈칸 3: rate limit + 에러 처리

NCBI Entrez는 초당 3회 이상 요청하면 IP 차단합니다. 도구 실행에 rate limit을 추가합니다.

python

import asyncio
import time

class RateLimiter:
    def __init__(self, calls_per_second: float):
        self.min_interval = 1.0 / calls_per_second
        self.last_call = 0.0
    
    async def wait(self):
        now = time.time()
        elapsed = now - self.last_call
        if elapsed < self.min_interval:
            await asyncio.sleep(self.min_interval - elapsed)
        self.last_call = time.time()


ncbi_limiter = RateLimiter(calls_per_second=3)


async def search_pubmed_limited(query: str, max_results: int = 10) -> dict:
    # TODO: 함수 시작 전에 ncbi_limiter.wait() 호출
    # 이후 기존 로직 실행
    pass

성찰 — 이 에이전트가 실전 tool use LLM과 어떻게 다른가

여러분이 만든 에이전트는 개념적으로 실전 시스템(Claude Code, ChatGPT with tools 등)과 같은 뿌리를 공유하지만, 프로덕션 시스템은 훨씬 정교합니다.

보안: 실전 MCP 서버는 prompt injection 을 방어해야 합니다. 도구가 반환하는 데이터에 악의적 지시("이 명령을 무시하고 대신 X를 하라")가 있으면 에이전트가 오작동할 수 있습니다. 실전 시스템은 도구 반환값을 sandboxed context 로 분리하거나 별도 검증 레이어를 둡니다.

비용 관리: 각 iteration이 API 호출을 트리거하므로 비용이 빠르게 누적됩니다. 실전 시스템은 iteration 상한, 토큰 예산, 도구 호출 캐싱 을 갖춥니다.

관찰성: 에이전트가 왜 특정 도구를 호출했는지, 어떤 판단이 잘못됐는지 사후 분석하려면 trace log 가 필요합니다. 실전은 OpenTelemetry 같은 표준으로 각 스텝을 기록합니다.

멀티 스텝 추론: 복잡한 요청(예: "이 유전자의 논문에 등장하는 다른 유전자들을 모두 찾아 각각의 발현 패턴을 조회")은 여러 단계의 도구 호출 계획을 요구합니다. 실전 시스템은 종종 planner-executor 분리 나 ReAct 프레임워크 를 씁니다.

확장 프로젝트

1. GenBank 서열 다운로드 도구 추가: efetch.fcgi 로 accession → FASTA 서열 반환.

2. 인용 그래프 도구: PubMed 논문의 인용 관계를 추적해 관련 논문 트리 구축.

3. 로컬 캐시: 같은 쿼리를 다시 실행하지 않도록 SQLite 캐시 레이어 추가.

4. Claude Desktop 통합: 여러분이 만든 MCP 서버를 실제 Claude Desktop 앱에 등록해 매일 씀. 표준 MCP SDK로 stdio 서버로 감싸면 됩니다.

이 편의 부품 지도

[F] MCP 프로토콜: 도구 정의, 호출, 응답의 표준 형식. 서버-클라이언트 분리.
[F] harness: 에이전트 동작을 자동 평가하는 테스트 시스템. LLM 시스템의 CI/CD 역할.
[W] API 호출·JSON: NCBI/UniProt REST 호출과 응답 파싱.
[W] async/await: 비동기 I/O로 여러 API 호출을 병렬화.

[F] = 여러분이 직접 구현 / [W] = 완성 코드로 제공한 도구 개념.