MCP Server

Pipe incluye un MCP (Model Context Protocol) server que permite a los agentes de IA gestionar y ejecutar pipelines de datos. Cualquier cliente compatible con MCP — como Claude Desktop — puede listar, ejecutar y monitorear pipelines a traves de lenguaje natural.

Instalar

pip install dataspoc-pipe[mcp]

Iniciar el servidor

dataspoc-pipe mcp

El servidor se ejecuta en stdio, siguiendo el protocolo de transporte MCP. Esta disenado para ser iniciado por un cliente MCP, no para ejecutarse manualmente en una terminal.

Configuracion de Claude Desktop

Agrega lo siguiente al archivo de configuracion MCP de Claude Desktop:

macOS: ~/Library/Application Support/Claude/claude_desktop_config.json

Linux: ~/.config/Claude/claude_desktop_config.json

{
  "mcpServers": {
    "dataspoc-pipe": {
      "command": "dataspoc-pipe",
      "args": ["mcp"]
    }
  }
}

Si dataspoc-pipe esta instalado en un entorno virtual, usa la ruta completa:

{
  "mcpServers": {
    "dataspoc-pipe": {
      "command": "/home/you/.venv/bin/dataspoc-pipe",
      "args": ["mcp"]
    }
  }
}

Reinicia Claude Desktop despues de actualizar la configuracion.

Herramientas disponibles

El MCP server expone las siguientes herramientas:

list_pipelines

Lista todos los nombres de pipelines configurados.

Retorna: Un nombre de pipeline por linea, o “No pipelines configured.”

pipeline_config

Retorna la configuracion completa de un pipeline como JSON.

Parametros:

• name (string, requerido): Nombre del pipeline

Retorna: JSON con configuracion de source, destination, incremental y schedule.

run_pipeline

Ejecuta un pipeline de extraccion.

Parametros:

• name (string, requerido): Nombre del pipeline
• full (boolean, opcional): Forzar extraccion completa, ignorando el estado incremental. Por defecto: false

Retorna: JSON con success, streams (conteo de registros por stream) y error.

pipeline_status

Retorna el estado de todos los pipelines configurados.

Retorna: Array JSON, cada entrada con name, last_run, status, duration, records.

pipeline_logs

Retorna el log de ejecucion mas reciente de un pipeline.

Parametros:

• name (string, requerido): Nombre del pipeline

Retorna: JSON con el log completo de ejecucion, o “No logs found”.

show_manifest

Retorna el manifiesto (catalogo) de un bucket.

Parametros:

• bucket (string, requerido): URI del bucket (ej., s3://my-bucket, file:///tmp/lake)

Retorna: JSON con el catalogo de tablas incluyendo esquemas, marcas de tiempo y conteos de filas.

validate_pipeline

Valida la conectividad del bucket y la disponibilidad del tap para un pipeline.

Parametros:

• name (string, requerido): Nombre del pipeline

Retorna: JSON con pipeline, bucket_ok, tap_ok y errors.

Recursos

pipe://pipelines

Un recurso MCP que lista todos los nombres de pipelines con su tap y bucket:

[
  {"name": "orders", "tap": "tap-csv", "bucket": "s3://my-lake"},
  {"name": "customers", "tap": "tap-postgres", "bucket": "s3://my-lake"}
]

Ejemplos de interaccion con agentes

Una vez configurado, puedes interactuar con Pipe a traves de Claude usando lenguaje natural:

“Que pipelines estan configurados?”

El agente llama a list_pipelines y retorna la lista.

“Ejecuta el pipeline de orders”

El agente llama a run_pipeline(name="orders") y reporta: “El pipeline de orders se completo exitosamente. Se extrajeron 5,200 registros en 1 stream.”

“Muestrame el estado de todos los pipelines”

El agente llama a pipeline_status y presenta un resumen formateado del ultimo run, estado, duracion y conteo de registros de cada pipeline.

“Haz una re-extraccion completa de customers”

El agente llama a run_pipeline(name="customers", full=True) para ignorar el estado incremental y re-extraer todo.

“El pipeline de orders esta saludable? Verifica el bucket y el tap.”

El agente llama a validate_pipeline(name="orders") y reporta si el bucket es escribible y si el tap esta disponible.

“Que tablas hay en el bucket s3://my-lake?”

El agente llama a show_manifest(bucket="s3://my-lake") y lista las tablas disponibles con sus esquemas y conteos de registros.