> ## Documentation Index
> Fetch the complete documentation index at: https://clickhouse.com/docs/llms.txt
> Use this file to discover all available pages before exploring further.

# StreamlitでClickHouseをバックエンドにしたAIエージェントを構築する方法

> StreamlitとClickHouse MCPサーバーを使ってWebベースのAIエージェントを構築する方法を学びます

このガイドでは、[Streamlit](https://streamlit.io/) を使って Web ベースの AI エージェントを構築し、[ClickHouseのMCPサーバー](https://github.com/ClickHouse/mcp-clickhouse) と [Agno](https://github.com/agno-agi/agno) を介して [ClickHouseのSQL Playground](https://sql.clickhouse.com/) と連携させる方法を学びます。

<Info>
  **サンプルアプリケーション**

  この例では、ClickHouse のデータをクエリできるチャットインターフェイスを備えた Web アプリケーションを構築します。
  この例のソースコードは、[examples リポジトリ](https://github.com/ClickHouse/examples/tree/main/ai/mcp/streamlit) にあります。
</Info>

<div id="prerequisites">
  ## 前提条件
</div>

* システムに Python がインストールされている必要があります。
  [`uv`](https://docs.astral.sh/uv/getting-started/installation/) がインストールされている必要があります。
* Anthropic の API キー、または別の LLM プロバイダの API キーが必要です

以下の手順を実行して、Streamlit アプリケーションを作成できます。

<Steps>
  <Step title="ライブラリのインストール" id="install-libraries">
    以下のコマンドを実行して、必要なライブラリをインストールします。

    ```bash theme={null}
    pip install streamlit agno ipywidgets
    ```
  </Step>

  <Step title="ユーティリティファイルを作成する" id="create-utilities">
    2 つのユーティリティ関数を含む `utils.py` ファイルを作成します。1 つ目は、
    Agno エージェントからのストリーム応答を処理するための非同期ジェネレーター関数です。
    2 つ目は、Streamlit
    アプリケーションにスタイルを適用するための関数です。

    ```python title="utils.py" theme={null}
    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)
    ```
  </Step>

  <Step title="認証情報の設定" id="setup-credentials">
    Anthropic APIキーを環境変数として設定します。

    ```bash theme={null}
    export ANTHROPIC_API_KEY="your_api_key_here"
    ```

    <Info>
      **別のLLMプロバイダーを使用する**

      AnthropicのAPIキーをお持ちでなく、別のLLMプロバイダーを使用したい場合は、
      [Agno「Integrations」ドキュメント](https://docs.agentops.ai/v2/integrations/ag2)で認証情報の設定手順を確認できます
    </Info>
  </Step>

  <Step title="必要なライブラリをインポートする" id="import-libraries">
    まず、メインのStreamlitアプリケーションファイル (例: `app.py`) を作成し、必要なライブラリをインポートします。

    ```python theme={null}
    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
    ```
  </Step>

  <Step title="agent のストリーミング関数を定義する" id="define-agent-function">
    [ClickHouse's SQL Playground](https://sql.clickhouse.com/) に接続し、レスポンスをストリーミングするメインの agent 関数を追加します。

    ```python theme={null}
    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
    ```
  </Step>

  <Step title="同期ラッパー関数を追加する" id="add-wrapper-functions">
    Streamlit での非同期ストリーミングを処理するためのヘルパー関数を追加します。

    ```python theme={null}
    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)
    ```
  </Step>

  <Step title="Streamlit インターフェイスを作成する" id="create-interface">
    Streamlit の UI コンポーネントとチャット機能を追加します。

    ```python theme={null}
    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})
    ```
  </Step>

  <Step title="アプリケーションを実行する" id="run-application">
    ClickHouse AI agent のWebアプリケーションを起動するには、ターミナルで
    次のコマンドを実行します。

    ```bash theme={null}
    uv run \
      --with streamlit \
      --with agno \
      --with anthropic \
      --with mcp \
      streamlit run app.py --server.headless true
    ```

    これにより Web ブラウザーが開き、`http://localhost:8501` に移動します。そこで
    AI エージェントと対話し、ClickHouse の SQL Playground で利用できる
    サンプルデータセットについて質問できます。
  </Step>
</Steps>