귀하의 MCP 서버는 API 어댑터가 아닙니다
Source: Dev.to
위에 제공된 소스 링크 외에 번역할 텍스트가 포함되어 있지 않습니다. 번역이 필요한 전체 내용을 제공해 주시면 한국어로 번역해 드리겠습니다.
많은 MCP 서버가 이렇게 보입니다:
@mcp.tool()
async def get_thing(id: str):
resp = await httpx.get(f"https://api.example.com/things/{id}")
return resp.json()가져오고, 전달하고, 끝. 얇은 HTTP 프록시에 JSON Schema 래퍼를 씌운 형태죠. 일부 사용 사례에서는 이것만으로 충분합니다.
제가 계속 다시 찾게 되는 서버들은 뭔가 다릅니다. 상태를 보관하고 답을 미리 계산해 둡니다. 얇은 래퍼에 접근하는 에이전트는 세 번의 왕복과 30 초가 필요할 수 있지만, 실제 작업을 수행하는 서버에 접근하면 한 번의 호출로 1 밀리초 이내에 답을 얻을 수 있습니다.
미리 로드된 인‑메모리 인덱스
제가 자주 마주치는 실패 상황이 있습니다: 에이전트가 무언가를 찾아야 하는데 정확한 ID를 모르는 경우입니다. 대부분의 API는 정확한 조회만 지원합니다. ID가 없으면 결과가 없습니다. 대화는 “그 리소스를 찾을 수 없습니다”라는 메시지와 함께 막히고, 사용자는 포기하게 됩니다.
저는 CDN 관리 API를 감싸는 서버를 만들었습니다. 수백 개의 프로퍼티가 있고, 에이전트는 종종 특정 호스트명을 처리하는 프로퍼티를 찾아야 합니다. API에 검색 엔드포인트가 있지만, 느리고 정확한 매치만 지원하며, 계정 권한에 따라 403을 반환하기도 합니다.
그래서 서버는 시작 시 모든 프로퍼티를 메모리로 로드합니다:
class PropertyIndex:
_entries: list[PropertyEntry] = field(default_factory=list)
_name_index: dict[str, int] = field(default_factory=dict)
async def load(self, refresh_interval: int = 300):
await self._build_index()
self._refresh_task = asyncio.create_task(self._refresh_loop())
def search_by_name(self, query: str, limit: int = 50):
names = [e.property_name for e in self._entries]
matches = process.extract(
query,
names,
scorer=fuzz.WRatio,
limit=limit,
score_cutoff=50,
)
return [self._entries[idx].to_dict() for _, score, idx in matches]한 번만 구축하고 병렬 API 호출을 전개해 중복을 제거한 뒤, 백그라운드에서 5분마다 새로 고칩니다. 조회는 1 밀리초 미만에 완료됩니다.
이게 없으면 에이전트는 정확한 프로퍼티 이름을 추측하고, 잘못된 것을 선택해 재시도하면서 세 번의 턴을 소모합니다. 인덱스를 사용하면 사용자가 “checkout에 대한 CDN 설정”이라고 입력했을 때 첫 시도에서 바로 올바른 답을 얻습니다. 이런 차이가 사람들을 에이전트를 계속 사용하게 할지, 수동으로 돌아가게 할지를 결정합니다.
CI/CD 서버에서도 같은 방식을 적용했습니다. API는 ID로 빌드 구성을 가져올 수 있지만 퍼지 검색을 제공하지 않습니다. ID를 모르면 막히게 되죠. 서버는 시작 시 모든 빌드 구성을 캐시하고, 퍼지 매칭을 수행합니다. 에이전트가 “payments 서비스의 배포 작업을 찾아줘”라고 하면, CI 시스템 자체는 할 수 없는 일을 즉시 순위화된 리스트로 반환합니다.
내장형 분석 데이터베이스
또 다른 서버는 관계형 데이터베이스 앞에 위치합니다. 일부 테이블은 2천만 행에 달합니다. 에이전트는 “이 지역에서 가장 높은 볼륨을 가진 제공자는 누구인가?” 혹은 “특정 카테고리의 상위 퍼포머를 보여줘”와 같은 분석 질문에 답해야 합니다.
데이터베이스는 이러한 쿼리를 위해 설계되지 않았습니다. 웹 UI용으로 좁고 잘 인덱싱된 조회만을 위해 구축되었습니다. 에이전트의 접근 패턴은 다릅니다: 애플리케이션이 절대 조인하지 않는 테이블 간의 광범위한 분석 질문을 합니다. 인덱스를 추가하는 것도 옵션이 아니었습니다. 데이터베이스는 다른 팀이 소유하고 있었고, AI 에이전트의 쿼리 패턴에 맞게 최적화하는 것은 로드맵에 없었습니다. 이런 쿼리 중 일부는 읽기 복제본에서 10–30 초가 걸렸으며, 에이전트 루프에서는 툴 호출 횟수만큼 지연이 곱해져 대화가 타임아웃됩니다.
서버는 프로세스 내에 DuckDB를 임베드하고, 시작 시 사전 집계된 뷰와 조회 테이블을 로드합니다. 일부는 작은 참조 테이블을 그대로 복사한 것이고, 다른 일부는 원본 데이터베이스가 효율적으로 실행하도록 설계되지 않은 조인을 평탄화한 물리화된 요약본입니다.
e kind of cross‑table aggregations that make sense for an analytical question but would be expensive on a schema built for transactional web UI lookups:
class DuckDBCache:
async def start(self):
self._conn = duckdb.connect(":memory:")
for key, config in fast_configs.items():
await self._load_table(key, config)
self._ready = True
self._deferred_task = asyncio.create_task(
self._load_deferred(deferred_configs)
)
self._refresh_task = asyncio.create_task(self._refresh_loop())각 테이블에는 지문 쿼리(저렴한 COUNT(*) 또는 체크섬)가 있어, 전체 재로드를 수행하기 전에 리프레시 루프가 이를 확인합니다. 큰 테이블은 서버가 이미 요청을 받고 있는 상태에서 백그라운드로 로드됩니다. 아직 로드되지 않은 테이블을 요청하면 소스 데이터베이스로 대체됩니다.
30초 걸리던 쿼리가 이제 1밀리초 이하로 실행됩니다. 에이전트가 첫 번째 질문 후 타임아웃되지 않고 사용자와 실제로 왕복 대화를 할 수 있게 됩니다.
이 위에 쿼리 결과 캐시도 있습니다. 사전 워밍 매니페스트(시작 시 실행되는 일반 쿼리 목록)를 가지고 있어, 월요일 아침에 에이전트를 처음 사용하는 사람이 콜드 스타트 없이 바로 사용할 수 있습니다.
class QueryCache:
async def get_or_compute(self, cache_key, compute_fn, ttl=None):
cached = self.get(cache_key)
if cached is not None:
return cached
result = await compute_fn()
if "error" not in result:
self._put(cache_key, result, ttl or self._default_ttl)
return result오류 응답은 캐시하지 않습니다. 데이터베이스가 일시적으로 과부하되어 쿼리가 실패하면, 그 실패를 다음 요청에 제공하고 싶지 않기 때문입니다.
Source: …
우리의
그것을 알아내는 데 생산 중단이 필요했습니다.
데이터 변환
내가 구축하는 모든 서버는 상위 API 응답을 반환하기 전에 정제합니다. 토큰 사용량은 응답 크기에 비례하고, 대부분의 API는 에이전트가 실제로 볼 데이터보다 10배 더 많은 데이터를 반환합니다.
내가 작업하는 한 API는 60개 이상의 필드를 가진 객체를 반환합니다. 서버는 8개 정도만 유지합니다:
def _slim_record(r: dict):
return _strip_nulls({
"id": r.get("id"),
"name": r.get("name"),
"total_value": _cents_to_major(r.get("total_value_cents")),
"annual_value": _cents_to_major(r.get("annual_value_cents")),
"start_date": r.get("start_date"),
"end_date": r.get("end_date"),
"status": _effective_status(r),
})_cents_to_major는 센트를 달러로 변환합니다. 원시 API는 금액을 센트 단위로 저장합니다.
이 변환을 추가하기 전에는 **100 %**의 보고서에서 에이전트가 잘못된 숫자를 보여주었습니다. 모든 금액이 100배씩 차이났습니다. $2,000 계약이 $200,000으로 표시된 이유는 에이전트가 센트를 달러로 취급했기 때문입니다. 프롬프트 엔지니어링만으로는 신뢰할 수 있게 고칠 수 없었습니다. 변환을 서버로 옮기니 해결되었습니다.
_effective_status도 언급할 가치가 있습니다. API의 상태 필드는 **“active”**라고 표시될 수 있지만, 실제로는 3개월 전에 종료된 레코드일 수 있습니다. 플랫폼 자체 UI는 여러 필드를 조합해 실제 상태를 도출하므로, MCP 서버도 동일하게 처리합니다:
def _effective_status(r: dict) -> str:
stage = r.get("stage")
if stage in ("terminated", "not_renewed"):
return "inactive"
if r.get("end_date_not_applicable") or r.get("renewal_type") == "perpetual":
return r.get("status", "undetermined")
end_date = r.get("end_date")
if end_date:
if date.fromisoformat(end_date) str:
return self.content[self._offsets[start - 1] : self._line_end_offset(end)]나는 이를 CDN 엣지‑함수 코드 번들 및 PDF 문서( PyMuPDF 로 추출) 에 사용합니다. 첫 번째 다운로드 후, 에이전트는 라인 범위로 읽고, 정규식으로 검색하고, 파일 트리를 나열합니다. 재다운로드가 없습니다. 200페이지 분량 문서를 읽는 것이 “다운로드, 추출, 읽기”가 아니라 “그냥 읽기”가 됩니다.
얇은 구조가 괜찮은 경우
모든 것이 이런 처리를 필요로 하는 것은 아닙니다. 자연어를 쿼리 언어로 변환해 API에 전달하는 서버는 얇은 래퍼로 충분합니다—그 변환 자체가 가치이기 때문입니다. 간단한 조회 도구도 마찬가지입니다.
스스로에게 물어볼 질문
- 에이전트가 동일한 데이터를 두 번 조회하나요?
- API가 에이전트가 필요로 하는 것보다 더 많은 데이터를 반환하나요?
- API 응답 시간이 느려서 에이전트 루프가 끊어진 느낌이 나나요?
예라면, 서버가 작업을 수행해야 합니다.
곱셈 효과
사람이 웹 UI를 사용할 때는 페이지를 보고, 생각하고, 다른 것을 클릭합니다—한 번에 하나의 요청을 인간 두뇌가 처리합니다. 에이전트는 다르게 동작합니다. 다섯 번의 도구 호출을 하고, 그 다섯 응답을 모두 컨텍스트 윈도우에 넣어 한 번에 추론합니다.
- 느린 응답은 호출마다 곱해집니다.
- 60필드 JSON 블롭도 호출마다 곱해집니다.
그 결과는 빠르게 누적됩니다.
측정된 차이
- CDN 속성 조회가 퍼지 인덱스를 적용한 뒤 세 번의 에이전트 턴에서 한 번으로 줄었습니다.
- 분석 쿼리가 30초 타임아웃에서 DuckDB를 사용해 1밀리초 이하로 반환되었습니다.
- 모든 보고서의 모든 금액이 서버가 센트를 달러로 변환하기 전까지는 잘못되었습니다.
마지막 경우를 프롬프트 엔지니어링으로 해결하려고 시도할 수 있습니다. 저는 몇 주 동안 시도했지만, 에이전트가 여전히 자주 틀려서 출력을 신뢰할 수 없었습니다.