Comment créer un agent IA adossé à ClickHouse avec Streamlit

Dans ce guide, vous apprendrez à créer un agent IA web avec Streamlit capable d’interagir avec le SQL playground de ClickHouse à l’aide du serveur MCP de ClickHouse et d’Agno.

Prérequis

Python doit être installé sur votre système. uv doit également être installé
Vous aurez besoin d'une API key Anthropic, ou d'une API key d'un autre fournisseur de LLM

Vous pouvez suivre les étapes ci-dessous pour créer votre application Streamlit.

Installer les bibliothèques

Installez les bibliothèques requises en exécutant les commandes suivantes :

pip install streamlit agno ipywidgets

Créer le fichier utilitaire

Créez un fichier utils.py contenant deux fonctions utilitaires. La première est un générateur asynchrone destiné à gérer les réponses en flux de l’agent Agno. La seconde est une fonction permettant d’appliquer des styles à l’application Streamlit :

utils.pypython

import streamlit as st
from agno.run.response import RunEvent, RunResponse

async def as_stream(response):
    async for chunk in response:
        if isinstance(chunk, RunResponse) and isinstance(chunk.content, str):
            if chunk.event == RunEvent.run_response:
                yield chunk.content

def apply_styles():
    st.markdown("""
  <style>
  hr.divider {
  background-color: white;
  margin: 0;
  }
  </style>
  <hr class='divider' />""", unsafe_allow_html=True)

Configurer les identifiants

Définissez votre clé API Anthropic comme variable d’environnement :

export ANTHROPIC_API_KEY="your_api_key_here"

Importez les bibliothèques nécessaires

Commencez par créer le fichier principal de votre application Streamlit (par ex. app.py) et ajoutez les importations :

from utils import apply_styles

import streamlit as st
from textwrap import dedent

from agno.models.anthropic import Claude
from agno.agent import Agent
from agno.tools.mcp import MCPTools
from agno.storage.json import JsonStorage
from agno.run.response import RunEvent, RunResponse
from mcp.client.stdio import stdio_client, StdioServerParameters

from mcp import ClientSession

import asyncio
import threading
from queue import Queue

Définir la fonction de streaming de l’agent

Ajoutez la fonction principale de l’agent qui se connecte au SQL playground de ClickHouse et diffuse les réponses en continu :

async def stream_clickhouse_agent(message):
    env = {
            "CLICKHOUSE_HOST": "sql-clickhouse.clickhouse.com",
            "CLICKHOUSE_PORT": "8443",
            "CLICKHOUSE_USER": "demo",
            "CLICKHOUSE_PASSWORD": "",
            "CLICKHOUSE_SECURE": "true"
        }
    
    server_params = StdioServerParameters(
        command="uv",
        args=[
        'run',
        '--with', 'mcp-clickhouse',
        '--python', '3.13',
        'mcp-clickhouse'
        ],
        env=env
    )
    
    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            mcp_tools = MCPTools(timeout_seconds=60, session=session)
            await mcp_tools.initialize()
            agent = Agent(
                model=Claude(id="claude-3-5-sonnet-20240620"),
                tools=[mcp_tools],
                instructions=dedent("""\
                    You are a ClickHouse assistant. Help users query and understand data using ClickHouse.
                    - Run SQL queries using the ClickHouse MCP tool
                    - Present results in markdown tables when relevant
                    - Keep output concise, useful, and well-formatted
                """),
                markdown=True,
                show_tool_calls=True,
                storage=JsonStorage(dir_path="tmp/team_sessions_json"),
                add_datetime_to_instructions=True, 
                add_history_to_messages=True,
            )
            chunks = await agent.arun(message, stream=True)
            async for chunk in chunks:
                if isinstance(chunk, RunResponse) and chunk.event == RunEvent.run_response:
                    yield chunk.content

Ajouter des fonctions wrapper synchrones

Ajoutez des fonctions utilitaires pour gérer le streaming asynchrone dans Streamlit :

def run_agent_query_sync(message):
    queue = Queue()
    def run():
        asyncio.run(_agent_stream_to_queue(message, queue))
        queue.put(None)  # Sentinel to end stream
    threading.Thread(target=run, daemon=True).start()
    while True:
        chunk = queue.get()
        if chunk is None:
            break
        yield chunk

async def _agent_stream_to_queue(message, queue):
    async for chunk in stream_clickhouse_agent(message):
        queue.put(chunk)

Créer l’interface Streamlit

Ajoutez les composants de l’interface Streamlit ainsi que la fonctionnalité de chat :

st.title("A ClickHouse-backed AI agent")

if st.button("💬 New Chat"):
  st.session_state.messages = []
  st.rerun()

apply_styles()

if "messages" not in st.session_state:
  st.session_state.messages = []

for message in st.session_state.messages:
  with st.chat_message(message["role"]):
    st.markdown(message["content"])

if prompt := st.chat_input("What is up?"):
  st.session_state.messages.append({"role": "user", "content": prompt})
  with st.chat_message("user"):
    st.markdown(prompt)
  with st.chat_message("assistant"):
    response = st.write_stream(run_agent_query_sync(prompt))
  st.session_state.messages.append({"role": "assistant", "content": response})

Lancez l’application

Pour démarrer votre application web d’agent IA ClickHouse, vous pouvez exécuter la commande suivante dans votre terminal :

uv run \
  --with streamlit \
  --with agno \
  --with anthropic \
  --with mcp \
  streamlit run app.py --server.headless true

Cela ouvrira votre navigateur web et vous redirigera vers http://localhost:8501, où vous pourrez interagir avec votre agent d’IA et lui poser des questions sur les jeux de données d’exemple disponibles dans le SQL playground de ClickHouse.