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

# ClickHouse MCP 서버를 사용하여 OpenAI 에이전트를 빌드하는 방법

> ClickHouse MCP 서버와 상호 작용할 수 있는 OpenAI 에이전트를 빌드하는 방법을 알아봅니다.

이 가이드에서는 [OpenAI](https://github.com/openai/openai-agents-python) 에이전트를 빌드하여
[ClickHouse SQL playground](https://sql.clickhouse.com/)에서 [ClickHouse MCP 서버](https://github.com/ClickHouse/mcp-clickhouse)와 상호 작용하는 방법을 알아봅니다.

<Info>
  **예시 노트북**

  이 예시는 [examples 리포지토리](https://github.com/ClickHouse/examples/blob/main/ai/mcp/openai-agents/openai-agents.ipynb)의 노트북에서 확인할 수 있습니다.
</Info>

<div id="prerequisites">
  ## 사전 요구 사항
</div>

* 시스템에 Python이 설치되어 있어야 합니다.
* 시스템에 `pip`이 설치되어 있어야 합니다.
* OpenAI API Key가 필요합니다.

다음 단계는 Python REPL이나 스크립트에서 실행할 수 있습니다.

<Steps>
  <Step title="라이브러리 설치" id="install-libraries">
    다음 명령을 실행하여 필요한 라이브러리를 설치하십시오:

    ```python theme={null}
    pip install -q --upgrade pip
    pip install -q openai-agents
    ```
  </Step>

  <Step title="자격 증명 설정" id="setup-credentials">
    다음으로 OpenAI API Key를 입력해야 합니다:

    ```python theme={null}
    import os, getpass
    os.environ["OPENAI_API_KEY"] = getpass.getpass("Enter OpenAI API Key:")
    ```

    ```response title="Response" theme={null}
    Enter OpenAI API Key: ········
    ```
  </Step>

  <Step title="MCP Server 및 OpenAI 에이전트 초기화" id="initialize-mcp-and-agent">
    이제 ClickHouse MCP 서버가 ClickHouse SQL 플레이그라운드를 가리키도록 구성하고,
    OpenAI 에이전트를 초기화한 다음 질문을 입력하십시오:

    ```python theme={null}
    from agents.mcp import MCPServer, MCPServerStdio
    from agents import Agent, Runner, trace
    import json

    def simple_render_chunk(chunk):
        """Simple version that just filters important events"""

        # Tool calls
        if (hasattr(chunk, 'type') and
                chunk.type == 'run_item_stream_event'):

            if chunk.name == 'tool_called':
                tool_name = chunk.item.raw_item.name
                args = chunk.item.raw_item.arguments
                print(f"🔧 Tool: {tool_name}({args})")

            elif chunk.name == 'tool_output':
                try:
                    # Handle both string and already-parsed output
                    if isinstance(chunk.item.output, str):
                        output = json.loads(chunk.item.output)
                    else:
                        output = chunk.item.output

                    # Handle both dict and list formats
                    if isinstance(output, dict):
                        if output.get('type') == 'text':
                            text = output['text']
                            if 'Error' in text:
                                print(f"❌ Error: {text}")
                            else:
                                print(f"✅ Result: {text[:100]}...")
                    elif isinstance(output, list) and len(output) > 0:
                        # Handle list format
                        first_item = output[0]
                        if isinstance(first_item, dict) and first_item.get('type') == 'text':
                            text = first_item['text']
                            if 'Error' in text:
                                print(f"❌ Error: {text}")
                            else:
                                print(f"✅ Result: {text[:100]}...")
                    else:
                        # Fallback - just print the raw output
                        print(f"✅ Result: {str(output)[:100]}...")

                except (json.JSONDecodeError, AttributeError, KeyError) as e:
                    # Fallback to raw output if parsing fails
                    print(f"✅ Result: {str(chunk.item.output)[:100]}...")

            elif chunk.name == 'message_output_created':
                try:
                    content = chunk.item.raw_item.content
                    if content and len(content) > 0:
                        print(f"💬 Response: {content[0].text}")
                except (AttributeError, IndexError):
                    print(f"💬 Response: {str(chunk.item)[:100]}...")

        # Text deltas for streaming
        elif (hasattr(chunk, 'type') and
              chunk.type == 'raw_response_event' and
              hasattr(chunk, 'data') and
              hasattr(chunk.data, 'type') and
              chunk.data.type == 'response.output_text.delta'):
            print(chunk.data.delta, end='', flush=True)

    async with MCPServerStdio(
            name="ClickHouse SQL Playground",
            params={
                "command": "uv",
                "args": [
                    'run',
                    '--with', 'mcp-clickhouse',
                    '--python', '3.13',
                    'mcp-clickhouse'
                ],
                "env": env
            }, client_session_timeout_seconds = 60
    ) as server:
        agent = Agent(
            name="Assistant",
            instructions="Use the tools to query ClickHouse and answer questions based on those files.",
            mcp_servers=[server],
        )

        message = "What's the biggest GitHub project so far in 2025?"
        print(f"\n\nRunning: {message}")
        with trace("Biggest project workflow"):
            result = Runner.run_streamed(starting_agent=agent, input=message, max_turns=20)
            async for chunk in result.stream_events():
                simple_render_chunk(chunk)
    ```

    ```response title="Response" theme={null}
    Running: What's the biggest GitHub project so far in 2025?
    🔧 Tool: list_databases({})
    ✅ Result: amazon
    bluesky
    country
    covid
    default
    dns
    environmental
    food
    forex
    geo
    git
    github
    hackernews
    imdb
    log...
    🔧 Tool: list_tables({"database":"github"})
    ✅ Result: {
      "database": "github",
      "name": "actors_per_repo",
      "comment": "",
      "columns": [
        {
          "...
    🔧 Tool: run_select_query({"query":"SELECT repo_name, MAX(stars) FROM github.top_repos_mv"})
    ✅ Result: {
      "status": "error",
      "message": "Query failed: HTTPDriver for https://sql-clickhouse.clickhouse....
    🔧 Tool: run_select_query({"query":"SELECT repo_name, stars FROM github.top_repos ORDER BY stars DESC LIMIT 1"})
    ✅ Result: {
      "repo_name": "sindresorhus/awesome",
      "stars": 402893
    }...
    The biggest GitHub project in 2025, based on stars, is "[sindresorhus/awesome](https://github.com/sindresorhus/awesome)" with 402,893 stars.💬 Response: The biggest GitHub project in 2025, based on stars, is "[sindresorhus/awesome](https://github.com/sindresorhus/awesome)" with 402,893 stars.
    ```
  </Step>
</Steps>