> ## 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 서버를 사용해 웹 기반 AI 에이전트를 구축하는 방법을 알아보세요

이 가이드에서는 [Streamlit](https://streamlit.io/)을 사용해 웹 기반 AI 에이전트를 구축하고, [ClickHouse의 SQL 플레이그라운드](https://sql.clickhouse.com/)와 [ClickHouse MCP 서버](https://github.com/ClickHouse/mcp-clickhouse), [Agno](https://github.com/agno-agi/agno)를 통해 상호작용하는 방법을 알아봅니다.

<Info>
  **예시 애플리케이션**

  이 예시는 ClickHouse 데이터를 쿼리할 수 있는 채팅 인터페이스를 제공하는 완전한 웹 애플리케이션을 생성합니다.
  이 예시의 소스 코드는 [예시 리포지토리](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 Key 또는 다른 LLM 제공업체의 API Key가 필요합니다.

다음 단계에 따라 Streamlit 애플리케이션을 만들 수 있습니다.

<Steps>
  <Step title="라이브러리 설치" id="install-libraries">
    다음 명령을 실행해 필요한 라이브러리를 설치합니다.

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

  <Step title="유틸리티 파일 생성" id="create-utilities">
    두 개의 유틸리티 함수를 담은 `utils.py` 파일을 만드십시오. 첫 번째는
    Agno agent의 스트림 응답을 처리하는 비동기 함수 생성기이고, 두 번째는
    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 Key를 환경 변수로 설정하세요:

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

    <Info>
      **다른 LLM 제공업체 사용**

      Anthropic API Key가 없고 다른 LLM 제공업체를 사용하려는 경우,
      [Agno "Integrations" 문서](https://docs.agentops.ai/v2/integrations/ag2)에서 자격 증명 설정 방법을 확인할 수 있습니다.
    </Info>
  </Step>

  <Step title="필요한 라이브러리 가져오기" id="import-libraries">
    먼저 기본 Streamlit 애플리케이션 파일(예: `app.py`)을 만들고 import 문을 추가합니다:

    ```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 웹 애플리케이션을 시작하려면 터미널에서 다음 명령을 실행하십시오:

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

    웹 브라우저가 열리고 `http://localhost:8501`로 이동합니다. 여기에서
    AI 에이전트와 상호작용하며 ClickHouse의 SQL Playground에서 사용할 수 있는 예시 데이터셋에 대해
    질문할 수 있습니다.
  </Step>
</Steps>