Home
Resources
Community

[8' read]

View on GitHub ↗Open in Colab ↗

Build a Database Advisor Agent with the DeepWiki Connector (Python)

ConnectorsAgents

You need a database for write-heavy local analytics. SQLite, DuckDB, and LevelDB are all strong contenders, but which one actually fits? Rather than reading documentation by hand, this notebook lets Mistral read their actual source code via the DeepWiki Connector and decide.

This notebook demonstrates the full Mistral Connector lifecycle:

StepOperationWhat happens
1CreateRegister a connector for each database candidate
2ListVerify all three are registered
3UseBuild an agent that compares them via their GitHub repos
4UpdateMark the winner's connector as selected
5DeleteClean up the losing connectors

API status: This notebook uses client.beta.connectors and client.beta.agents. These are beta endpoints and may change.

Run cells top-to-bottom. A TypeScript version of the same agent is also available here.

Prerequisites#

To complete this notebook, you will need:

  • Python 3.9 or later
  • A Mistral account and API key

Environment setup#

Install the Mistral Python SDK by running the cell below.

To complete this cookbook, you'll need a Mistral API key. In Studio, navigate to the API keys section, choose Private and shared connectors for Connector access scope and create a new API key.

Set it before running the client cell using one of these options:

Option 1 — environment variable (recommended for local use):

MISTRAL_API_KEY=your-mistral-api-key

Option 2 — enter it when prompted: if MISTRAL_API_KEY is not already set in your environment, the next code cell will display a secure input field where you can paste your key directly.

%pip install mistralai --quiet

Import the SDK and create the client. If MISTRAL_API_KEY is not set as an environment variable, a secure input prompt will appear.

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"])

Step 1 — Create one connector per candidate#

Each connector points at the DeepWiki MCP server, which lets Mistral read and reason about any public GitHub repository. We create three named Connectors (one per candidate) so each one acts as a named slot the agent can query independently.

After creating each connector, create_or_update_user_credentials registers credentials for it. DeepWiki is a public server that requires no authentication, so credentials is an empty dict — but the credentials record must still exist before the connector can be queried.

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}")

Step 2 — List to verify#

Confirm all three connectors are registered, then call list_tools on each one to verify the credentials are working and see what tools the connector exposes.

View your registered Connectors in 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}")

Step 3 — Build the comparison agent#

We create a Mistral agent with all three connectors attached. The agent's instructions do two things:

  1. Direct the agent to use the connectors — it must ask each DeepWiki connector a natural-language question about the repository before forming an opinion. Asking questions rather than reading raw source files keeps the responses concise enough to fit in the context window.

  2. Require JSON output — the agent must return a single JSON object matching the schema defined below. The schema is defined in code so it serves as a precise, readable contract for the expected structure. Note that the conversations API does not support response_format alongside agent_id, so JSON conformance is enforced via the agent's instructions rather than at the API level.

Response schema#

We define the expected JSON structure before creating the agent. The schema has four required fields:

  • queries — one entry per database, capturing the question sent to each connector and the summary returned. This makes the connector calls visible in the output rather than hidden inside the model's reasoning.
  • comparison — a fixed set of evaluation dimensions (storage_model, acid_guarantees, query_capabilities, write_throughput, python_api), each as a brief comparative string.
  • reasoning — a single paragraph explaining the final choice.
  • recommendation — one of the three connector names, constrained to an enum so the value is always a valid key in connectors.

Defining the schema in code (rather than describing it in the instruction string) means it serves as the single source of truth: the same object is referenced in the agent's instructions and passed to the API as a json_schema response format, so the structure is enforced at the API level rather than relying on the model to follow prose directions.

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"],
}

Create the agent with the three connectors attached. The completion_args field passes RESPONSE_SCHEMA to the API as a json_schema response format, so the model is constrained to return valid JSON matching the schema on every run.

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})")

Step 4 — Run the comparison#

This step uses two calls intentionally. start_stream_async keeps the connection open so you can watch connector calls happen in real time — tool.execution.started shows which database is being queried, and tool.execution.done confirms the result came back. But the streaming API sends all agent turn outputs as identical message.output.delta events with no "is final" flag, so you can't reliably extract the json_schema-constrained response from the stream directly.

get_messages_async solves this cleanly. It returns structured MessageOutputEntry objects for the completed conversation, so the last assistant message is unambiguously the final JSON answer. Because the agent may produce multiple outputs across turns (for example, an intermediate summary before the final answer), all concatenated in a single string, raw_decode parses one JSON object at a time — iterating to the end gives the last one, which is always the schema-constrained answer.

import json

conversation_id = None
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,
):
    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":
        name = getattr(data, "name", "")
        print(f"[{event_type}]{' ' + name if name else ''}")

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', '')}")

Because the agent returned structured JSON, extracting the winner is a direct key lookup — no regex required. We also validate that the recommendation is one of the known connector names before proceeding.

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)}")

Step 5 — Promote the winner, retire the rest#

Update the winning connector's description to mark it as selected, then delete the losing connectors. This completes the full lifecycle: 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 connector is still registered with its updated description.

# 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#

Delete the agent when you're done. Uncomment the last two lines to also remove the winning 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#

This notebook demonstrated the full Mistral Connector lifecycle — create, list, use, update, and delete — using the DeepWiki Connector to let the model read actual GitHub repository source code and produce a data-driven database recommendation.

What you built:

  • Three named Connectors pointing at the DeepWiki MCP server
  • An agent (Database Showdown Judge) with all three Connectors attached
  • A conversation that produced a structured recommendation, updated the winner's Connector, and cleaned up the rest

Mistral features used:

  • Connectors (beta)
  • Agents API (beta)
  • Conversations API (beta)

Other services:

  • DeepWiki — MCP server for reading public GitHub repositories

View your Connectors in Studio.