DeepWiki 커넥터로 데이터베이스 어드바이저 에이전트 만들기
DeepWiki 커넥터로 데이터베이스 어드바이저 에이전트 만들기 (Build a Database Advisor Agent with the DeepWiki Connector, Python)
쓰기 부하가 많은 로컬 분석에 쓸 데이터베이스가 필요하다고 가정해 볼게요. SQLite, DuckDB, LevelDB 모두 강력한 후보지만, 실제로 어느 것이 맞을까요? 문서를 손으로 읽는 대신, 이 노트북은 Mistral이 DeepWiki 커넥터를 통해 실제 소스 코드를 읽고 결정하게 해줘요.
출처: 문서
본문
쓰기 부하가 많은 로컬 분석을 위한 데이터베이스가 필요하다고 가정해 볼게요. SQLite, DuckDB, LevelDB 모두 강력한 후보지만, 실제로 어느 것이 맞을까요? 문서를 손으로 읽기보다는, 이 노트북에서는 Mistral이 DeepWiki 커넥터를 통해 그들의 실제 소스 코드를 읽고 결정하게 해요.
이 노트북은 전체 Mistral Connector 수명주기를 보여줘요.
| Step | Operation | What happens |
|---|---|---|
| 1 | Create | Register a connector for each database candidate |
| 2 | List | Verify all three are registered |
| 3 | Use | Build an agent that compares them via their GitHub repos |
| 4 | Update | Mark the winner's connector as selected |
| 5 | Delete | Clean up the losing connectors |
API 상태 (API status): 이 노트북은 client.beta.connectors와 client.beta.agents를 사용해요. 이들은 베타(beta) 엔드포인트로 변경될 수 있어요.
셀을 위에서 아래로 실행하세요. 같은 에이전트의 TypeScript 버전도 여기에서 사용할 수 있어요.
사전 준비사항 (Prerequisites)
이 노트북을 완료하려면 다음이 필요해요.
- Python 3.9 이상
- Mistral 계정과 API 키
환경 설정 (Environment setup)
아래 셀을 실행해서 Mistral Python SDK를 설치해요.
이 쿡북을 완료하려면 Mistral API 키가 필요해요. Studio에서 API keys 섹션으로 이동해서, Connector access scope에 대해 Private and shared connectors를 선택하고 새 API 키를 만들어요.
클라이언트 셀을 실행하기 전에 다음 옵션 중 하나로 설정하세요.
옵션 1 — 환경 변수 (로컬 사용에 권장):
MISTRAL_API_KEY=your-mistral-api-key
옵션 2 — 프롬프트 때 입력: MISTRAL_API_KEY가 환경에 아직 설정되어 있지 않다면, 다음 코드 셀에서 키를 직접 붙여넣을 수 있는 안전한 입력 필드가 표시돼요.
%pip install mistralai --quiet
SDK를 가져오고 클라이언트를 만들어요. MISTRAL_API_KEY가 환경 변수로 설정되어 있지 않다면, 안전한 입력 프롬프트가 나타나요.
import getpass
import os
from mistralai.client import Mistral
if not os.environ.get("MISTRAL_API_KEY"):
os.environ["MISTRAL_API_KEY"] = getpass.getpass("Mistral API key: ")
client = Mistral(api_key=os.environ["MISTRAL_API_KEY"])
1단계 — 후보마다 커넥터 하나 생성 (Step 1 — Create one connector per candidate)
각 커넥터는 DeepWiki MCP 서버를 가리켜요. 이 서버 덕분에 Mistral이 공개 GitHub 저장소를 읽고 추론할 수 있어요. 우리는 후보당 하나씩 세 개의 이름 있는 Connector를 만들어서, 각각이 에이전트가 독립적으로 쿼리할 수 있는 이름 있는 슬롯 역할을 하게 해요.
각 커넥터를 만든 후 create_or_update_user_credentials로 자격 증명을 등록해요. DeepWiki는 인증이 필요 없는 공개 서버라 credentials는 빈 dict이지만, 커넥터를 쿼리하려면 자격 증명 레코드가 여전히 존재해야 해요.
DEEPWIKI_URL = "https://mcp.deepwiki.com/mcp"
candidates = [
{"name": "showdown_sqlite", "description": "DeepWiki connector — sqlite/sqlite"},
{"name": "showdown_duckdb", "description": "DeepWiki connector — duckdb/duckdb"},
{"name": "showdown_leveldb", "description": "DeepWiki connector — google/leveldb"},
]
connectors = {}
for c in candidates:
connector = await client.beta.connectors.create_async(
name=c["name"],
description=c["description"],
server=DEEPWIKI_URL,
visibility="private",
)
connectors[c["name"]] = connector
print(f"Created: {connector.name} (id={connector.id})")
await client.beta.connectors.create_or_update_user_credentials_async(
connector_id_or_name=connector.name,
name=f"{connector.name}-default",
credentials={"headers": {}},
is_default=True,
)
print(f" Credentials registered for {connector.name}")
2단계 — 목록으로 확인 (Step 2 — List to verify)
세 커넥터가 모두 등록됐는지 확인한 다음, 각각에 list_tools를 호출해서 자격 증명이 작동하는지 확인하고 커넥터가 노출하는 도구를 살펴보아요.
등록된 Connector를 Studio에서 확인할 수 있어요.
page = await client.beta.connectors.list_async(page_size=50)
showdown = [c for c in page.items if c.name.startswith("showdown_")]
print(f"{len(showdown)} showdown connectors registered:\n")
for c in showdown:
print(f" {c.name:<22} {c.description}")
tools = await client.beta.connectors.list_tools_async(
connector_id_or_name=c.name,
)
for tool in tools:
print(f" - {tool.name}: {tool.description}")
3단계 — 비교 에이전트 만들기 (Step 3 — Build the comparison agent)
세 개의 커넥터를 모두 연결한 Mistral 에이전트를 만들어요. 에이전트의 지시사항은 두 가지를 합니다.
커넥터를 사용하도록 에이전트를 지시 — 의견을 형성하기 전에 각 DeepWiki 커넥터에 저장소에 대한 자연어 질문을 해야 해요. 원시 소스 파일을 읽는 대신 질문을 하면 응답을 컨텍스트 창에 들어갈 만큼 간결하게 유지할 수 있어요.
JSON 출력을 요구 — 에이전트는 아래 정의된 스키마와 일치하는 단일 JSON 객체를 반환해야 해요. 스키마는 코드로 정의되므로 예상 구조에 대한 정확하고 읽기 쉬운 계약 역할을 해요. conversations API는 agent_id와 함께 response_format을 지원하지 않으므로, JSON 준수는 API 수준이 아니라 에이전트의 지시사항을 통해 강제돼요.
응답 스키마 (Response schema)
에이전트를 만들기 전에 예상 JSON 구조를 정의해요. 스키마에는 네 개의 필수 필드가 있어요.
queries— 데이터베이스당 하나씩, 각 커넥터에 보낸 질문과 반환된 요약을 담아요. 이렇게 하면 커넥터 호출이 모델의 추론 안에 숨지 않고 출력에서 보이게 해요.comparison— 고정된 평가 차원(storage_model,acid_guarantees,query_capabilities,write_throughput,python_api)으로, 각각 간략한 비교 문자열이에요.reasoning— 최종 선택을 설명하는 단일 문단이에요.recommendation— 세 커넥터 이름 중 하나로,enum으로 제한되어 값이 항상connectors의 유효한 키가 되도록 해요.
스키마를 (지시 문자열로 설명하는 대신) 코드로 정의하면 단일 소스가 돼요. 같은 객체가 에이전트의 지시사항에서 참조되고 API에 json_schema 응답 형식으로 전달되므로, 모델이 산문 지시를 따르는 데 의존하지 않고 구조가 API 수준에서 강제돼요.
RESPONSE_SCHEMA = {
"type": "object",
"properties": {
"queries": {
"type": "array",
"items": {
"type": "object",
"properties": {
"connector": {"type": "string"},
"question": {"type": "string"},
"summary": {"type": "string"},
},
"required": ["connector", "question", "summary"],
},
},
"comparison": {
"type": "object",
"properties": {
"storage_model": {"type": "string"},
"acid_guarantees": {"type": "string"},
"query_capabilities": {"type": "string"},
"write_throughput": {"type": "string"},
"python_api": {"type": "string"},
},
"required": ["storage_model", "acid_guarantees", "query_capabilities", "write_throughput", "python_api"],
},
"reasoning": {"type": "string"},
"recommendation": {"type": "string", "enum": ["showdown_sqlite", "showdown_duckdb", "showdown_leveldb"]},
},
"required": ["queries", "comparison", "reasoning", "recommendation"],
}
세 개의 커넥터를 연결한 에이전트를 만들어요. completion_args 필드는 RESPONSE_SCHEMA를 API에 json_schema 응답 형식으로 전달해서, 모델이 매 실행마다 스키마와 일치하는 유효한 JSON을 반환하도록 제한해요.
from mistralai.client.models import CompletionArgs, JSONSchema, ResponseFormat
agent = await client.beta.agents.create_async(
name="Database Showdown Judge",
description="Compares database candidates using their source code via DeepWiki.",
model="mistral-medium-latest",
instructions=(
"You are a database selection expert. "
"When given a comparison task, call each DeepWiki connector once with a focused "
"natural-language question about the repository — do NOT read raw source files."
),
completion_args=CompletionArgs(
response_format=ResponseFormat(
type="json_schema",
json_schema=JSONSchema(
name="comparison_result",
schema_definition=RESPONSE_SCHEMA,
strict=True,
),
),
),
tools=[
{"type": "connector", "connector_id": connectors["showdown_sqlite"].id},
{"type": "connector", "connector_id": connectors["showdown_duckdb"].id},
{"type": "connector", "connector_id": connectors["showdown_leveldb"].id},
],
)
print(f"Agent ready: {agent.name} (id={agent.id})")
4단계 — 비교 실행 (Step 4 — Run the comparison)
이 단계는 의도적으로 두 개의 호출을 사용해요. start_stream_async는 연결을 열어서 커넥터 호출이 실시간으로 일어나는 것을 볼 수 있게 해줘요 — tool.execution.started는 어떤 데이터베이스가 쿼리 중인지 보여주고, tool.execution.done은 결과가 돌아왔음을 확인해줘요. 하지만 스트리밍 API는 모든 에이전트 턴 출력을 "is final" 플래그 없이 동일한 message.output.delta 이벤트로 보내기 때문에, 스트림에서 직접 json_schema로 제한된 응답을 안정적으로 추출할 수 없어요.
get_messages_async가 이 문제를 깔끔하게 해결해요. 완료된 대화에 대한 구조화된 MessageOutputEntry 객체를 반환하므로, 마지막 어시스턴트 메시지가 명확하게 최종 JSON 답변입니다. 에이전트는 여러 턴에 걸쳐 여러 출력을 만들 수 있으므로(예: 최종 답변 전 중간 요약), 모두 단일 문자열로 연결되어요. raw_decode는 한 번에 하나의 JSON 객체를 파싱해서 — 끝까지 반복하면 마지막 것(항상 스키마로 제한된 답변)을 얻어요.
import json
import asyncio
import time
conversation_id = None
last_event_time = time.monotonic()
async for event in await client.beta.conversations.start_stream_async(
agent_id=agent.id,
inputs=[
{
"role": "user",
"content": (
"Compare sqlite/sqlite, duckdb/duckdb, and google/leveldb for a write-heavy "
"local analytics workload. Evaluate storage model, ACID guarantees, query "
"capabilities, write throughput, and Python API simplicity. Recommend one."
),
}
],
timeout_ms=300_000,
):
last_event_time = time.monotonic()
data = event.data
event_type = getattr(data, "type", None)
if event_type == "conversation.response.started":
conversation_id = data.conversation_id
elif event_type == "message.output.delta":
print(".", end="", flush=True) # progress indicator
else:
name = getattr(data, "name", "")
print(f"\n[{event_type}]{' ' + name if name else ''}")
print(f"\n\nStream finished — fetching messages...")
messages = await client.beta.conversations.get_messages_async(
conversation_id=conversation_id
)
last_output = next(
(m for m in reversed(messages.messages) if getattr(m, "type", None) == "message.output"),
None,
)
content = last_output.content
raw_text = content if isinstance(content, str) else "".join(
getattr(c, "text", "") for c in content
)
# raw_decode handles concatenated JSON — returns the last object, which is the final answer.
decoder = json.JSONDecoder()
result, pos = None, 0
while pos < len(raw_text):
try:
result, pos = decoder.raw_decode(raw_text, pos)
while pos < len(raw_text) and raw_text[pos] in " \n\r\t":
pos += 1
except json.JSONDecodeError:
pos += 1
print("\n--- Connector queries ---")
for q in result.get("queries", []):
print(f"\n [{q['connector']}]")
print(f" Q: {q['question']}")
print(f" A: {q['summary']}")
print("\n--- Comparison ---")
for key, val in result.get("comparison", {}).items():
print(f" {key}: {val}")
print(f"\n--- Reasoning ---\n {result.get('reasoning', '')}")
print(f"\n--- Recommendation ---\n {result.get('recommendation', '')}")
에이전트가 구조화된 JSON을 반환했기 때문에 승자를 추출하는 것은 regex 없이 직접 키 조회일 뿐이에요. 또한 진행 전에 추천이 알려진 커넥터 이름 중 하나인지 검증해요.
winner_name = result.get("recommendation", "").strip()
if winner_name not in connectors:
raise ValueError(f"Unexpected recommendation {winner_name!r} — expected one of {list(connectors)}")
loser_names = [name for name in connectors if name != winner_name]
print(f"Winner: {winner_name}")
print(f"Losers: {', '.join(loser_names)}")
5단계 — 승자 승격, 나머지 은퇴 (Step 5 — Promote the winner, retire the rest)
승리한 커넥터의 설명을 업데이트해서 선택됨을 표시한 다음, 진 커넥터들을 삭제해요. 이로써 전체 수명주기(create → list → use → update → delete)가 완료돼요.
# Mark the winner
winner_connector = connectors[winner_name]
updated = await client.beta.connectors.update_async(
connector_id=winner_connector.id,
description=f"[SELECTED] {winner_connector.description}",
)
print(f"Updated: {updated.name} — {updated.description}")
# Delete the losers
for name in loser_names:
delete_result = await client.beta.connectors.delete_async(connector_id=connectors[name].id)
print(f"Deleted: {name} — {delete_result.message}")
승자 커넥터가 업데이트된 설명과 함께 여전히 등록되어 있는지 확인해요.
# Confirm the winner is still there with its updated description
winner = await client.beta.connectors.get_async(connector_id_or_name=winner_name)
print("Winner confirmed:")
print(f" Name: {winner.name}")
print(f" Description: {winner.description}")
print(f" ID: {winner.id}")
정리 (Cleanup)
작업이 끝나면 에이전트를 삭제해요. 마지막 두 줄의 주석을 해제하면 승리한 Connector도 제거돼요.
await client.beta.agents.delete_async(agent_id=agent.id)
print(f"Agent deleted: {agent.id}")
# Uncomment to also remove the winning Connector:
# result = await client.beta.connectors.delete_async(connector_id=winner_connector.id)
# print(f"Connector deleted: {winner_name}")
요약 (Summary)
이 노트북은 DeepWiki Connector를 사용해 모델이 실제 GitHub 저장소 소스 코드를 읽고 데이터 기반의 데이터베이스 추천을 내도록 하면서, Mistral Connector의 전체 수명주기(create, list, use, update, delete)를 보여줬어요.
만든 것 (What you built):
- DeepWiki MCP 서버를 가리키는 세 개의 이름 있는 Connector
- 세 개의 Connector가 모두 연결된 에이전트(Database Showdown Judge)
- 구조화된 추천을 만들고, 승자의 Connector를 업데이트하고, 나머지를 정리한 대화
사용한 Mistral 기능 (Mistral features used):
- Connectors (beta)
- Agents API (beta)
- Conversations API (beta)
기타 서비스 (Other services):
- DeepWiki — 공개 GitHub 저장소를 읽기 위한 MCP 서버
Connector를 Studio에서 확인할 수 있어요.