Cómo crear un agente de OpenAI con el servidor MCP de ClickHouse.

Aprende a crear un agente de OpenAI que pueda interactuar con el servidor MCP de ClickHouse.

En esta guía, aprenderás a crear un agente de OpenAI que pueda interactuar con el Playground SQL de ClickHouse mediante el servidor MCP de ClickHouse.

Notebook de ejemplo

Este ejemplo está disponible como notebook en el repositorio de ejemplos.

​

Requisitos previos

• Deberás tener Python instalado en tu sistema.
• Deberás tener pip instalado en tu sistema.
• Deberás tener una API key de OpenAI

Puedes seguir los pasos siguientes desde el REPL de Python o mediante un script.

1. ​

   Instalar bibliotecas

   Instale la biblioteca necesaria con los siguientes comandos:

   pip install -q --upgrade pip
   pip install -q openai-agents
   
2. ​

   Configurar credenciales

   A continuación, deberás proporcionar tu clave de API de OpenAI:

   import os, getpass
   os.environ["OPENAI_API_KEY"] = getpass.getpass("Enter OpenAI API Key:")
   

   Response

   Enter OpenAI API Key: ········
   
3. ​

   Inicializar el servidor MCP y el agente de OpenAI

   Ahora configure el ClickHouse MCP server para que apunte al playground de ClickHouse SQL, inicialice su agente de OpenAI y hágale una pregunta:

   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"""
   
       # Llamadas a herramientas
       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:
                   # Manejar tanto cadenas como salida ya analizada
                   if isinstance(chunk.item.output, str):
                       output = json.loads(chunk.item.output)
                   else:
                       output = chunk.item.output
   
                   # Manejar tanto formato de diccionario como de lista
                   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:
                       # Manejar formato de lista
                       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 - imprimir la salida sin procesar
                       print(f"✅ Result: {str(output)[:100]}...")
   
               except (json.JSONDecodeError, AttributeError, KeyError) as e:
                   # Fallback a la salida sin procesar si el análisis falla
                   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]}...")
   
       # Deltas de texto para 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

   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.