Grabba MCP Server

Grabba MCP服务器

该存储库包含Grabba微服务连接器协议（MCP）服务器，旨在将Grabba API功能作为一组可调用工具公开。建立在 FastMCP，此服务器允许AI代理、编排器（如LangChain）和其他应用程序与Grabba数据提取和管理服务无缝交互。

目录

1. 特性
2. 入门指南

- 先决条件 - 安装 - 通过PyPI（推荐） - 来源（发展） - 运行服务器 - 局部的，局部的 - - 公共实例

1. 配置

- 环境变量 - 命令行参数

1. 可用工具

- 认证 - 工具详细信息

1. 连接到MCP服务器

- Python客户端（LangChain示例） - 可流式HTTP传输 -

1. 开发说明

- 项目结构 - 运行测试

1. 链接和资源
2. 许可证

______________________________________________________________________

特性

• 格拉巴API风险敞口： 将Grabba API的关键功能（数据提取、作业管理、统计）作为可访问的工具进行展示。
• 多种运输方式： 支持 stdio, streamable-http，以及 sse 传输，为不同的部署和客户端场景提供灵活性。
• 依赖注入： 利用FastAPI强大的依赖注入实现安全高效 GrabbaService 初始化（例如处理API密钥）。
• 容器化部署： 针对Docker进行了优化，便于打包和部署。
• 可配置： 允许通过环境变量和命令行参数进行配置。

______________________________________________________________________

入门指南

先决条件

• Python 3.10+
• Docker（用于容器化部署）
• Grabba API密钥（您可以从 Grabba网站)

安装

通过PyPI（推荐）

这 grabba-mcp PyPI上提供了该包。这是最简单的开始方式。

pip install grabba-mcp

来源（发展）

如果您计划贡献或修改服务器，则需要从源代码安装。

1. 克隆存储库：

   git clone https://github.com/grabba-dev/grabba-mcp
   cd grabba-mcp

1. 安装诗歌：

如果您没有安装Poetry，请遵循他们的官方指南：

   pip install poetry

1. 安装项目依赖关系：

导航到 apps/mcp 目录，其中 pyproject.toml 驻留，然后安装：

   cd apps/mcp
   poetry install

运行服务器

局部的，局部的

安装后（通过 pip 或者从源代码），您可以运行服务器。

1. 创建一个 .env 文件：

在 apps/mcp 目录（如果从源代码运行）或执行 grabba-mcp 命令，创建 .env 文件并添加您的Grabba API密钥：

   API_KEY="YOUR_API_KEY_HERE"
   # Optional: configure the server port
   PORT=8283
   # Optional: configure the default transport (overridden by CLI)
   MCP_SERVER_TRANSPORT="streamable-http"

1. 执行服务器：

- 如果通过安装 pip:

     grabba-mcp

要通过命令行指定传输，请执行以下操作：

     grabba-mcp streamable-http

- 如果从源代码运行（使用Poetry）：

     cd apps/mcp
     poetry run python src/server.py

要通过命令行指定传输，请执行以下操作：

     poetry run python src/server.py stdio

您应该看到指示服务器正在启动并侦听指定端口的输出（例如。， http://0.0.0.0:8283)如果使用HTTP传输。请注意 stdio 传输将在单个请求/响应周期后退出，使其不适合持久服务。

Docker容器

Docker Hub上提供了一个预构建的Docker镜像，使部署变得简单。

1. 拉取图像：

   docker pull itsobaa/grabba-mcp:latest

1. 运行容器：

对于持久服务器，您通常会使用 streamable-http 运输和绘制港口地图。

   docker run -d \
     -p 8283:8283 \
     -e API_KEY="YOUR_API_KEY_HERE" \
     -e MCP_SERVER_TRANSPORT="streamable-http" \
     itsobaa/grabba-mcp:latest

您还可以使用 docker-compose 对于更复杂的设置：

   # docker-compose.yml
   version: '3.8'
   services:
     grabba-mcp:
       image: itsobaa/grabba-mcp:latest
       container_name: grabba-mcp
       environment:
         API_KEY: ${API_KEY} # Reads from a .env file next to docker-compose.yml
         MCP_SERVER_TRANSPORT: streamable-http
         PORT: 8283
       ports:
         - "8283:8283"
       healthcheck:
         test: ["CMD-SHELL", "curl -f http://localhost:8283/tools/openapi.json || exit 1"]
         interval: 10s
         timeout: 5s
         retries: 5

用一个 docker-compose.yml 文件，创建 .env 其旁边的文件（例如。， API_KEY="YOUR_API_KEY_HERE")并运行：

   docker-compose up -d

公共实例

Grabba MCP服务器可在以下网址公开访问：

• 网址： https://mcp.grabba.dev/
• 运输： 支持 sse 和 streamable-http.
• 身份验证： 需要一个 API_KEY 使用您的Grabba API密钥进行收割。

______________________________________________________________________

配置

服务器可以通过环境变量和命令行参数进行配置。

环境变量

• API_KEY （必需）：您的Grabba API密钥。这对于使用Grabba服务进行身份验证至关重要。
• PORT （可选，默认值： 8283)：MCP服务器的HTTP传输端口(streamable-http, sse)会听的。
• MCP_SERVER_TRANSPORT （可选，默认值： stdio)：MCP服务器的默认传输协议。可以是 stdio, streamable-http，或 sse.

命令行参数

服务器还接受一个位置命令行参数，该参数覆盖 MCP_SERVER_TRANSPORT:

grabba-mcp [transport_protocol]
# or for source: python src/server.py [transport_protocol]

• [transport_protocol]：可以 stdio, streamable-http，或 sse.

- 例子： grabba-mcp streamable-http

______________________________________________________________________

可用工具

Grabba MCP服务器公开了一套工具，这些工具封装了Grabba Python SDK的功能。

认证

对于 streamable-http 和 sse 传输时，通过包括 API_KEY 带有Grabba API密钥的HTTP头。 例子： API_KEY: YOUR_API_KEY_HERE

对于 stdio 运输 API_KEY 必须在以下环境中设置环境变量 grabba-mcp 由于此通信模式中没有HTTP标头，因此执行命令。

工具详细信息

extract_data

• 说明： 使用Grabba安排新的数据提取作业。适用于网络搜索任务。
• 输入： Job 对象（Pydantic模型）详细描述了提取任务。
• 输出： tuple[str, Optional[Dict]] -一条消息和 JobResult 作为一本词典。

schedule_existing_job

• 说明： 安排一个现有的Grabba作业立即运行。
• 输入： job_id （string）-现有作业的ID。
• 输出： tuple[str, Optional[Dict]] -一条消息和 JobResult 作为一本词典。

fetch_all_jobs

• 说明： 获取当前用户的所有Grabba作业。
• 输入： 没有。
• 输出： tuple[str, Optional[List[Job]]] -一条消息和一个列表 Job 物体。

fetch_specific_job

• 说明： 通过ID获取特定Grabba作业的详细信息。
• 输入： job_id （string）-作业的ID。
• 输出： tuple[str, Optional[Job]] -一条消息和 Job 对象。

delete_job

• 说明： 删除特定的Grabba作业。
• 输入： job_id （string）-要删除的作业的ID。
• 输出： tuple[str, None] -成功信息。

fetch_job_result

• 说明： 通过结果ID获取已完成的Grabba作业的结果。
• 输入： job_result_id （string）-作业结果的ID。
• 输出： tuple[str, Optional[Dict]] -消息和作业结果数据作为字典。

delete_job_result

• 说明： 删除已完成的Grabba作业的结果。
• 输入： job_result_id （string）-要删除的作业结果的ID。
• 输出： tuple[str, None] -成功信息。

fetch_stats_data

• 说明： 获取Grabba的使用统计数据和当前用户令牌余额。
• 输入： 没有。
• 输出： tuple[str, Optional[JobStats]] -一条消息和 JobStats 对象。

estimate_job_cost

• 说明： 在创建或调度之前估算Grabba作业的成本。
• 输入： Job 对象（Pydantic模型）详细描述了提取任务。
• 输出： tuple[str, Optional[Dict]] -一条消息和估计成本明细作为字典。

create_job

• 说明： 在Grabba中创建新的数据提取作业，而不立即安排其执行。
• 输入： Job 对象（Pydantic模型）详细描述了提取任务。
• 输出： tuple[str, Optional[Job]] -一条消息和创建的 Job 对象。

fetch_available_regions

• 说明： 获取可用于安排web数据提取的所有可用木偶（web代理）区域的列表。
• 输入： 没有。
• 输出： tuple[str, Optional[List[PuppetRegion]]] -一条消息和一个列表 PuppetRegion 物体。

______________________________________________________________________

连接到MCP服务器

这 MultiServerMCPClient 从 mcp.client 旨在连接到FastMCP服务器。

Python客户端（LangChain示例）

这个例子假设你有 mcp-client 安装的软件包（通常作为更大的LangChain/Agent设置的一部分），以及 grabba 和 pydantic.

import asyncio
import os
from typing import List, Dict, Optional
from langchain_core.tools import BaseTool, Tool
from mcp.models.mcp_server_config import McpServerConfig, McpServer
from mcp.client.transports.streamable_http import StreamableHttpConnection
from mcp.client.transports.stdio import StdioConnection
from mcp.client.multi_server_client import MultiServerMCPClient
from grabba import Job, JobStats, PuppetRegion # Import necessary Grabba Pydantic models
from dotenv import load_dotenv # For loading API key from .env

async def connect_and_use_mcp_tools(mcp_server_configs: List[McpServerConfig], api_key: Optional[str] = None) -> List[Tool]:
    """
    Connects to the MCP server(s), discovers its tools, and wraps them as LangChain Tools.
    Handles API key injection for HTTP connections.
    """
    try:
        mcp_client_config = {}
        for config in mcp_server_configs:
            # Pydantic V2 model validation
            mcp_server_model = McpServer.model_validate(config.mcp_server.model_dump())
            
            connection_headers = {}
            if api_key:
                # Use standard header name for API keys
                connection_headers["API_KEY"] = api_key 

            if mcp_server_model.transport == "streamable_http":
                server_params: StreamableHttpConnection = {
                    "transport": "streamable_http",
                    "url": str(mcp_server_model.url),
                    "env": config.env_variables or {}, # For other env variables, if any
                    "headers": connection_headers # Pass headers for HTTP transports
                }
            elif mcp_server_model.transport == "stdio":
                server_params: StdioConnection = {
                    "transport": "stdio",
                    "command": mcp_server_model.command, 
                    "args": mcp_server_model.args, 
                    "env": config.env_variables # For stdio, env maps to subprocess env vars
                }
            else:
                raise ValueError(f"Unsupported transport: {mcp_server_model.transport}")

            print(f"Client connecting with params: {server_params}")
            mcp_client_config[mcp_server_model.name] = server_params
        
        mcp_client = MultiServerMCPClient(mcp_client_config)
        tools: List[BaseTool] = await mcp_client.get_tools()
        print(f"Successfully loaded {len(tools)} tools.")
        return tools
    except Exception as e:
        print(f"Error connecting to MCP server or loading tools: {e}")
        return []

async def main():
    load_dotenv() # Load API key from a client-side .env file
    API_KEY = os.getenv("API_KEY", "YOUR_API_KEY_HERE_IF_NOT_ENV")

    # --- Configuration for Streamable HTTP Transport (Local or Public Instance) ---
    # For local: url="http://localhost:8283"
    # For public: url="https://mcp.grabba.dev/"
    http_mcp_config = McpServerConfig(
        mcp_server=McpServer(
            name="grabba-agent-http",
            transport="streamable_http",
            url="http://localhost:8283" # Or "https://mcp.grabba.dev/" for public
        )
    )

    print("\n--- Connecting via Streamable HTTP ---")
    http_tools = await connect_and_use_mcp_tools(
        mcp_server_configs=[http_mcp_config],
        api_key=API_KEY
    )

    if http_tools:
        print("\nAvailable HTTP Tools:")
        for tool in http_tools:
            print(f"- {tool.name}: {tool.description.split('.')[0]}.")
        
        # Example: Using the extract_data tool (adjust as per your Job Pydantic model)
        extract_tool = next((t for t in http_tools if t.name == "extract_data"), None)
        if extract_tool:
            print("\n--- Testing extract_data tool via HTTP ---")
            sample_job = Job(
                url="https://example.com/some-page",
                type="markdown", # or "pdf", "html" etc.
                parser="text-content",
                strategy="auto"
                # ... other required fields for Job
            )
            try:
                result_msg, result_data = await extract_tool.ainvoke({"extraction_data": sample_job})
                print(f"Extraction Result (HTTP): {result_msg}")
                if result_data:
                    print(f"Extraction Data (HTTP): {result_data.get('extracted_text', 'No text extracted')[:100]}...") # Print first 100 chars
            except Exception as e:
                print(f"Error calling extract_data via HTTP: {e}")
        else:
            print("extract_data tool not found in HTTP tools.")

        # Example: Using fetch_all_jobs tool
        fetch_jobs_tool = next((t for t in http_tools if t.name == "fetch_all_jobs"), None)
        if fetch_jobs_tool:
            print("\n--- Testing fetch_all_jobs tool via HTTP ---")
            try:
                result_msg, jobs_list = await fetch_jobs_tool.ainvoke({})
                print(f"Fetch Jobs Result (HTTP): {result_msg}")
                if jobs_list:
                    print(f"Fetched {len(jobs_list)} jobs.")
                    for job in jobs_list[:2]: # Print first 2 jobs
                        print(f"  - Job ID: {job.job_id}, URL: {job.url}")
            except Exception as e:
                print(f"Error calling fetch_all_jobs via HTTP: {e}")
        
        # Example: Using fetch_stats_data tool
        fetch_stats_tool = next((t for t in http_tools if t.name == "fetch_stats_data"), None)
        if fetch_stats_tool:
            print("\n--- Testing fetch_stats_data tool via HTTP ---")
            try:
                result_msg, stats_data = await fetch_stats_tool.ainvoke({})
                print(f"Fetch Stats Result (HTTP): {result_msg}")
                if stats_data:
                    print(f"Token Balance (HTTP): {stats_data.token_balance}")
                    print(f"Jobs Run (HTTP): {stats_data.jobs_run_count}")
            except Exception as e:
                print(f"Error calling fetch_stats_data via HTTP: {e}")

    # --- Configuration for Stdio Transport (e.g., to a Docker container running the server) ---
    # This assumes you have the 'itsobaa/grabba-mcp:latest' Docker image available.
    # The client launches a temporary Docker container for each tool call.
    stdio_mcp_config = McpServerConfig(
        mcp_server=McpServer(
            name="grabba-agent-stdio",
            transport="stdio",
            command="docker",
            args=[
                "run",
                "-i",          # Keep STDIN open for interactive communication
                "--rm",        # Remove container after exit
                "itsobaa/grabba-mcp:latest", # The Docker Hub image for Grabba MCP server
                "grabba-mcp", "stdio" # Command to run the server in stdio mode inside container
            ],
            env_variables={"API_KEY": API_KEY} # Pass API key as env var for stdio
        )
    )

    print("\n--- Connecting via Stdio (to Docker container as a subprocess) ---")
    stdio_tools = await connect_and_use_mcp_tools(
        mcp_server_configs=[stdio_mcp_config],
        api_key=API_KEY # Client might still pass for internal consistency, though env_variables is primary for stdio
    )

    if stdio_tools:
        print("\nAvailable Stdio Tools:")
        for tool in stdio_tools:
            print(f"- {tool.name}: {tool.description.split('.')[0]}.")
        
        # Example: Using the fetch_available_regions tool via Stdio
        fetch_regions_tool = next((t for t in stdio_tools if t.name == "fetch_available_regions"), None)
        if fetch_regions_tool:
            print("\n--- Testing fetch_available_regions tool via Stdio ---")
            try:
                result_msg, regions_list = await fetch_regions_tool.ainvoke({})
                print(f"Fetch Regions Result (Stdio): {result_msg}")
                if regions_list:
                    print(f"Fetched {len(regions_list)} regions.")
                    for region in regions_list[:3]: # Print first 3 regions
                        print(f"  - {region.display_name} ({region.code})")
            except Exception as e:
                print(f"Error calling fetch_available_regions via Stdio: {e}")
        else:
            print("fetch_available_regions tool not found in Stdio tools.")

if __name__ == "__main__":
    asyncio.run(main())

______________________________________________________________________

开发说明

项目结构

your_project_root/
├── src/
│   └── server.py               # Main FastMCP server application
├── .env                        # Environment variables for local development
├── pyproject.toml              # Poetry project configuration
└── poetry.lock                 # Poetry dependency lock file
├── Dockerfile                  # Docker build instructions for the server
├── docker-compose.yml          # Docker Compose configuration for local development/deployment
├── .dockerignore               # Files to ignore during Docker build
├── .env                        # Example .env for docker-compose (for API_KEY)
├── README.md                   # This documentation file
├── pyproject.toml              # Root pyproject.toml (if using monorepo structure)
├── poetry.lock                 # Root poetry.lock (if using monorepo structure)
├── src/                        # Source code (often for the root project if it's a monorepo)
├── tests/                      # Project tests
└── ... (other project files like dist, docs, tox.ini, project.json etc.)

运行测试

运行测试（根据您的配置 pyproject.toml):

poetry run pytest

______________________________________________________________________

链接和资源

• Grabba网站： https://www.grabba.dev/
• Grabba MCP服务器公共实例： https://mcp.grabba.dev/
• GitHub存储库：
• Docker Hub镜像：
• PyPI包： https://pypi.org/project/grabba-mcp/

______________________________________________________________________

许可证

本项目根据专有许可证获得许可。请查看 LICENSE 存储库根目录中的文件以获取完整详细信息。

______________________________________________________________________