이전 글에서는 AI 에이전트에게 Python을 안전하게 실행할 공간을 마련해주었습니다. 그건 에이전트가 작성한 코드가 어디서 실행되는지에 대한 답이었습니다. 이번 후속 글은 그에 못지않게 중요한 질문에 답합니다. 모델이 실제로 봐야 하는 데이터는 무엇인가?
대개 그 답은 우리가 지금 넘겨주고 있는 양보다 훨씬 적습니다.
순진한 패턴
에이전트를 업무 시스템에 연결하는 가장 뻔한 방법은 이렇습니다.
- 도구가 시스템에 질의하고 결과 집합을 JSON으로 반환한다.
- 그 행들이 모델의 컨텍스트에 들어간다.
- 모델이 그것을 읽고 답한다. “상위 5개는…”, “평균은…”.
행이 20개뿐인 데모에서는 이게 통합니다. 실제 데이터 앞에서는 세 가지 방식으로 무너집니다.
- 보안. 도구가 반환한 모든 행이 이제 대화 기록의 일부가 됩니다. 대화 기록은 저장되고, 로그로 남고, 이후 턴에 다시 재생되고, 모델 제공업체로 전송되고, 텔레메트리에 수집되며, 때로는 평가용으로 보관됩니다. 원래 시스템 밖으로 나갈 필요가 전혀 없었던 데이터가 이제 그 모든 곳에 복사된 셈입니다.
- 비용과 용량. 필드 10개짜리 행 하나는 대략 40~60토큰입니다. 5,000행이면 수십만 토큰이 됩니다. 컨텍스트 윈도우의 대부분 혹은 전부이고, 그걸 들고 다니는 모든 턴마다 비용을 지불하게 됩니다.
- 정확성. LLM은 계산하지 않습니다. 추정할 뿐입니다. 모델에게 숫자 5,000개를 더하거나 전분기 대비 변화로 그룹 순위를 매기라고 하면, 자신만만하지만 근사치일 뿐 틀린 답이 나옵니다. 같은 분석이 pandas로는 세 줄이면 정확하게 나옵니다.
결국 모델은 자기가 가장 못하는 일(대량 데이터에 대한 산술)을, 가장 적게 가진 자원(컨텍스트)을 써서 하게 되고, 그 과정에서 데이터를 사방에 흩뿌립니다.
패턴: 데이터는 제자리에, 모델은 핸들만
발상을 뒤집어 봅시다. 데이터를 만들어내는 도구는 데이터를 반환하지 않습니다. 서버 쪽에 아티팩트로 저장해두고, 참조와 함께 모델이 그걸 대상으로 코드를 작성하는 데 필요한 만큼의 설명만 반환합니다.
+--------------------------- MCP server ----------------------------+
| |
| fetch_* ---> [ artifact store ] <---> run_python (sandbox) |
| ^ ^ | |
| | +---- outputs ----+ |
| +---- render_chart / export |
| |
+------------------------------|------------------------------------+
| only: dataRef, schema, row count,
| tiny sample, small results
v
+---------------+
| LLM / agent |
+---------------+
모델의 역할이 “데이터를 읽고 답하기"에서 “데이터의 형태를 이해하고 답을 내는 코드를 작성하기"로 바뀝니다. 모델은 그 역할을 잘 해냅니다.
세 종류의 도구가 이 저장소를 사용합니다.
- 생산자(
fetch_*)는 호출자의 권한으로 쿼리를 실행하고, 결과를 저장한 뒤 설명을 반환합니다. - 계산 도구(
run_python)는 참조를 입력으로 받습니다. 샌드박스 안에서는 이것이 DataFrame으로 나타납니다. 스크립트가 저장하는 모든 DataFrame은 새로운 아티팩트가 되고, 모델에게는 작은 JSON 결과만 돌아갑니다. - 소비자(
render_chart,export_csv)는 참조를 받아 사람이 보는 형태로 바꿉니다. 행은 저장소에서 사용자 화면으로 바로 가고, 모델을 거치지 않습니다.
최소 구현
예제는 이전 글과 마찬가지로 FastMCP와 pandas를 사용합니다. 일부러 작게 만들었습니다.
저장소
import secrets
import time
from dataclasses import dataclass, field
import pandas as pd
@dataclass
class Artifact:
owner: str
df: pd.DataFrame
created: float = field(default_factory=time.monotonic)
class ArtifactStore:
def __init__(self, ttl_seconds: int = 24 * 3600, max_bytes: int = 50_000_000):
self._items: dict[str, Artifact] = {}
self._ttl = ttl_seconds
self._max_bytes = max_bytes
def put(self, owner: str, df: pd.DataFrame) -> str:
if df.memory_usage(deep=True).sum() > self._max_bytes:
raise ValueError("artifact exceeds size cap - narrow the query")
ref = "data_" + secrets.token_urlsafe(12)
self._items[ref] = Artifact(owner, df)
return ref
def get(self, owner: str, ref: str) -> pd.DataFrame:
now = time.monotonic()
self._items = {k: a for k, a in self._items.items() if now - a.created < self._ttl}
artifact = self._items.get(ref)
if artifact is None or artifact.owner != owner:
raise KeyError(f"unknown or expired dataRef {ref} - fetch the data again")
return artifact.df
여기서 중요한 세부사항이 두 가지 있습니다. 참조는 순차적이 아니라 무작위이고, 모든 읽기는 소유자를 확인합니다. 다른 사람의 대화 기록에 참조가 섞여 들어가도 그 사람에게는 쓸모가 없습니다. 에러 메시지는 에이전트에게 복구 방법도 알려줍니다.
쏟아내지 말고 설명하라
import json
def describe(ref: str, df: pd.DataFrame, sample_rows: int = 5) -> dict:
return {
"dataRef": ref,
"rows": len(df),
"columns": {c: str(t) for c, t in df.dtypes.items()},
"nulls": {c: int(n) for c, n in df.isna().sum().items() if n},
"sample": json.loads(df.head(sample_rows).to_json(orient="records", date_format="iso")),
}
컬럼 이름, 타입, null 개수, 그리고 몇 개의 샘플 행. 사람이 groupby를 작성하는 데 필요한 것이 이것이고, 모델에게 필요한 것도 똑같습니다.
생산자 도구와 계산 도구
from fastmcp import FastMCP, Context
mcp = FastMCP("data-sandbox")
store = ArtifactStore()
def owner_of(ctx: Context) -> str:
... # resolve the authenticated caller from your auth layer
@mcp.tool
async def fetch_sales(since: str, ctx: Context) -> dict:
"""Load sales since a date into a server-side artifact.
Returns a dataRef with schema and a small sample, NOT the rows.
Analyse the data with run_python."""
df = await query_sales(since=since) # your system of record, under the caller's permissions
return describe(store.put(owner_of(ctx), df), df)
@mcp.tool
async def run_python(code: str, inputs: dict[str, str], ctx: Context) -> dict:
"""Run Python in the sandbox. `inputs` maps variable names to dataRefs;
each arrives as a pandas DataFrame. Store DataFrames in `outputs[name]`
to keep them as new artifacts. Assign a SMALL JSON value to `result`."""
owner = owner_of(ctx)
with tempfile.TemporaryDirectory() as run:
in_dir, out_dir = Path(run, "in"), Path(run, "out")
in_dir.mkdir()
out_dir.mkdir()
for name, ref in inputs.items():
if not name.isidentifier():
raise ValueError(f"input name {name!r} must be a Python identifier")
store.get(owner, ref).to_parquet(in_dir / f"{name}.parquet")
res = await execute_sandboxed(code, in_dir, out_dir) # the sandbox from part one
new = {p.stem: pd.read_parquet(p) for p in out_dir.glob("*.parquet")}
outputs = {name: describe(store.put(owner, df), df, sample_rows=3) for name, df in new.items()}
return {"ok": res.ok, "result": cap_json(res.result), "stdout": truncate(res.stdout),
"error": res.error, "outputs": outputs}
샌드박스 안에서는 짧은 준비 코드가 각 in/*.parquet을 전역 변수로 불러오고 빈 outputs 딕셔너리를 만듭니다. 마무리 코드는 outputs를 out/에 기록하고 result를 직렬화합니다. 샌드박스는 여전히 네트워크가 없고 자기 실행 디렉터리만 볼 수 있으므로, 스크립트를 통해서도 데이터가 빠져나갈 수 없습니다. (런타임에 pyarrow가 없다면 CSV도 됩니다. 다만 컬럼 타입 정보는 잃게 됩니다.)
대화 기록은 이렇게 생겼다
user: Which five product lines lost the most revenue quarter over quarter?
agent -> fetch_sales(since="2025-01-01")
<- {dataRef: "data_x7Kq...", rows: 48210, columns: {...}, sample: [5 rows]}
agent -> run_python(inputs={"sales": "data_x7Kq..."}, code="""
q = sales.assign(q=sales.date.dt.to_period("Q"))
pivot = q.pivot_table(index="line", columns="q", values="revenue", aggfunc="sum")
delta = (pivot.iloc[:, -1] - pivot.iloc[:, -2]).sort_values()
outputs["qoq"] = delta.rename("delta").reset_index()
result = delta.head(5).round(2).to_dict()
""")
<- {result: {"Line A": -182340.5, ...}, outputs: {"qoq": {dataRef: "data_Pm2...", rows: 312}}}
agent: The five biggest drops were ...
... several turns later ...
user: Chart all of them.
agent -> render_chart(dataRef="data_Pm2...", x="line", y="delta", kind="bar")
48,210개 행이 관련되어 있었습니다. 대화 기록에 도달한 건 약 2킬로바이트였습니다. 답이 정확한 이유는 모델이 아니라 pandas가 계산했기 때문입니다.
단계와 턴을 넘나드는 아티팩트
참조는 짧은 문자열이기 때문에, 다른 값과 똑같이 대화 속을 돌아다닙니다.
- 연쇄되는 도구 호출. 한
run_python호출의outputs가 다음 호출의inputs가 됩니다. 여러 단계로 된 분석(정제, 조인, 집계, 순위 매기기)은 참조의 사슬입니다. 중간 테이블은 컨텍스트를 전혀 거치지 않습니다. - 나중 턴. 2번째 턴의 참조는 9번째 턴에도 대화 기록에 그대로 남아 있습니다. “이제 그걸 지역별로 나눠줘"는 저장된 아티팩트를 재사용합니다. 다시 가져올 필요도, 데이터를 다시 붙여넣을 필요도 없습니다.
- 만료도 설계의 일부. TTL이 저장소 크기를 일정하게 유지해줍니다. 만료된 참조는 다시 가져오라는 에러를 반환하므로, 오래된 핸들은 낡은 데이터를 조용히 돌려주는 대신 깔끔하게 실패합니다.
이 내용을 에이전트 지침에 적어두세요. 질문에 답하려고 원본 행을 요청하지 말 것. 아티팩트로 가져와서, run_python으로 계산하고, dataRef로 차트를 그릴 것.
이 패턴을 지탱하는 규칙
- 데이터의 내용이 아니라 형태를 반환하라. 스키마, 행 수, null 개수, 샘플 행 몇 개. 샘플 크기는 설정 가능하게 만들고, 민감한 데이터셋에는 0으로 설정하세요.
- 모델에게 돌아가는 모든 것을 작게 유지하고, 그걸 강제하라.
result의 크기를 제한하고stdout을 잘라내세요. 그러지 않으면print(df)한 줄이 테이블 전체를 모델에게 돌려보냅니다. - 참조는 각각 한 명의 소유자에게 속한다. 무작위 ID를 쓰고 읽을 때마다 소유자를 확인하세요. 참조는 권한 증표가 아닙니다.
- 생산자는 원본 시스템의 규칙을 지킨다. 아티팩트에는 호출자가 가져올 수 있도록 허용된 것만 정확히 담깁니다. 같은 인가, 같은 행 수 제한입니다.
- 저장소에 한계를 둬라. 아티팩트당 바이트 상한, TTL, 소유자 범위의 조회. 이곳은 작업용 데이터를 잠시 담아두는 곳이지, 데이터 웨어하우스가 아닙니다.
- 화면에 보여주는 모든 경로가 참조를 받도록 하라. 차트, 표, 내보내기 모두 저장소에서 직접 읽으므로, “보여줘” 단계에서도 행이 모델을 거치지 않습니다.
솔직한 트레이드오프
이 방식은 노출을 줄입니다. 없애지는 않습니다. 샘플 행과 집계값도 여전히 데이터입니다. 크기가 1인 그룹은 레코드 하나나 마찬가지이고, 작정하고 만든 프롬프트는 여전히 출력 상한 안에서 행을 출력하라고 에이전트에게 요구할 수 있습니다. 결과 상한과 샘플 크기는 정책 설정으로 다루고, 진짜 경계는 원본 시스템의 인가에 두세요.
모델은 이제 데이터를 직접 볼 수도 없습니다. 5행짜리 샘플은 오해를 부를 수 있습니다. 이상한 값, 섞여 있는 단위, 90%가 null인 컬럼처럼요. 요청 시 요약 통계를 반환하는 저렴한 describe_data(dataRef) 도구 하나면, 전체를 다시 쏟아내지 않고도 이런 문제 대부분을 해결할 수 있습니다.
마치며
1부의 샌드박스는 에이전트 코드 실행을 안전하게 만들었습니다. 데이터 아티팩트는 그걸 실행할 가치가 있게 만듭니다. 모델은 계획을 세우고 코드를 작성하고, Python이 산술을 하고, 데이터는 여러분의 시스템 안에 남습니다. 대화 기록에는 결국 질문과 코드와 답만 남습니다. 그 분석이 다룬 데이터는 하나도 담겨 있지 않은, 재현 가능한 분석 기록입니다.