Company MCP

数据库上下文MCP服务器

提供以下功能的FastMCP服务器 数据库模式上下文 通过搜索和元数据工具。此MCP提供预索引数据库文档——它确实 不 连接到实际的数据库。

目的

该服务器从JSON/PMarkdown文件中索引数据库模式（表、列、关系），并提供：

• FTS5全文搜索：BM25排名关键字搜索
• 向量相似性搜索：使用OpenAI嵌入的语义搜索
• 模式检索：从索引的JSON文件中获取表/列定义
• 关系发现：查找表之间的连接路径

数据结构

目录布局

data/
├── index/
│   └── index.db              # SQLite index with FTS5 + vector search
└── map/
    ├── postgres_production/
    │   └── domains/
    │       ├── authentication/
    │       │   └── tables/
    │       │       ├── auth.users.json
    │       │       ├── auth.users.md
    │       │       ├── auth.sessions.json
    │       │       └── ...
    │       ├── payments/
    │       │   └── tables/
    │       │       ├── public.payments.json
    │       │       ├── public.merchants.json
    │       │       └── ...
    │       ├── realtime/
    │       ├── security/
    │       └── storage/
    └── snowflake_production/
        └── domains/
            ├── payments/
            │   └── tables/
            │       └── PUBLIC.DABSTEP_PAYMENTS.json
            └── workflow/
                └── tables/
                    ├── PUBLIC.DABSTEP_TASKS.json
                    ├── PUBLIC.DABSTEP_SUBMISSIONS.json
                    └── PUBLIC.DABSTEP_TASK_SCORES.json

JSON模式文件格式

每个表都记录在具有以下结构的JSON文件中：

{
  "table": "payments",
  "schema": "public",
  "database": "postgres_production",
  "description": "Payment transactions processed through the system...",
  "row_count": 138236,
  "columns": [
    {
      "name": "payment_id",
      "type": "character varying",
      "nullable": false,
      "description": "Unique identifier for each payment transaction..."
    }
  ],
  "primary_key": ["payment_id"],
  "foreign_keys": [],
  "indexes": [
    {
      "index_name": "payments_pkey",
      "columns": ["payment_id"],
      "is_unique": true
    }
  ]
}

______________________________________________________________________

索引数据库架构

SQLite索引数据库(data/index/index.db)包含以下表格：

documents --主要文件索引

存储所有索引内容（表和列）。

documents_fts --FTS5全文搜索

BM25排名文本搜索的虚拟表。

CREATE VIRTUAL TABLE documents_fts USING fts5(
    content, summary, keywords,
    content='documents', content_rowid='id'
);

documents_vec --矢量嵌入

用于语义相似性搜索的虚拟表（需要sqlite-vec扩展）。

CREATE VIRTUAL TABLE documents_vec USING vec0(
    document_id INTEGER PRIMARY KEY,
    embedding float[1536]  -- OpenAI text-embedding-3-small
);

keywords --提取的搜索词

带有频率计数的缓存关键字。

index_weights --搜索排名配置

index_metadata --索引配置

元数据的键值存储，如 embedding_model, embedding_dimensions, document_count, last_full_index.

______________________________________________________________________

快速开始

1.安装依赖项

pip install -r requirements.txt

2.设置环境（可选，用于矢量搜索）

创建 .env 使用您的OpenAI API密钥文件：

OPENAI_API_KEY=sk-your-api-key-here

3.初始化数据库

python setup_db.py

这创造了 data/index/index.db 与：

• FTS5全文搜索索引
• 用于语义搜索的向量嵌入（如果提供了OpenAI密钥）
• 预先计算的关键字

4.运行服务器

python server.py

或者使用Docker：

docker build -t db-context-mcp .
docker run -p 8000:8000 --env-file .env db-context-mcp

MCP端点服务于 http://localhost:8000/mcp.

5.使用Python客户端进行测试

python client.py

______________________________________________________________________

Docker编写部署

对于完整部署 MCP服务器 和 SFTP服务器:

Docker Compose快速入门

# 1. Copy environment template
cp .env.example .env

# 2. Edit .env with your configuration
#    - Set a secure SFTP_PASSWORD
#    - Add OPENAI_API_KEY for vector search

# 3. Start all services
docker-compose up -d

服务

SFTP访问

连接到SFTP服务器以进行程序化文件访问：

# Connect via SFTP
sftp -P 2222 datauser@localhost

# Default credentials (change in .env):
# User: datauser
# Password: changeme

文件可在以下网址访问 /home/datauser/data/ 哪个映射到 ./data 目录。

环境变量

Docker命令

# Start all services
docker-compose up -d

# View logs
docker-compose logs -f

# Stop all services
docker-compose down

# Rebuild after changes
docker-compose up -d --build

# View service status
docker-compose ps

______________________________________________________________________

可用工具（共14个）

发现工具

list_databases

了解索引中可用的数据库。

list_databases()

退货：

{
  "databases": [
    {"name": "postgres_production", "table_count": 37, "domains": [...], "schemas": [...]},
    {"name": "snowflake_production", "table_count": 4, "domains": [...], "schemas": [...]}
  ]
}

list_domains

列出所有具有表计数的业务域。

list_domains(
    database: str = ""    # Optional: filter by database
)

退货： {domains: [{name, description, table_count, databases}]}

list_tables

列出带有元数据的可用表。

list_tables(
    database: str = "",   # Optional: filter by database
    domain: str = ""      # Optional: filter by domain
)

退货： 列表 {name, title, database, schema, domain, summary, file_path}

______________________________________________________________________

搜索工具

search_fts

使用SQLite FTS5进行全文搜索，BM25排名。

search_fts(
    query: str,           # Search text (supports AND, OR, NOT, "exact phrase")
    database: str = "",   # Optional: filter by database
    domain: str = "",     # Optional: filter by domain
    doc_type: str = "",   # Optional: filter by type ("table" or "column")
    limit: int = 10       # Max results (1-50)
)

退货： 结果与 id, table_name, summary, file_path, bm25_rank

search_vector

使用OpenAI嵌入的语义搜索。查找具有相似含义的文档。

search_vector(
    query: str,           # Natural language query
    database: str = "",   # Optional: filter by database
    domain: str = "",     # Optional: filter by domain
    doc_type: str = "",   # Optional: filter by type
    limit: int = 10       # Max results (1-50)
)

退货： 结果与 id, table_name, summary, file_path, distance

注： 需要 OPENAI_API_KEY 环境变量。

search_db_map

对内存表索引进行基于令牌的快速搜索。

search_db_map(
    query: str,           # Free-text search string
    top_k: int = 3        # Number of results
)

退货： 列表 {id, title, score, snippet}

search_tables

使用标记重叠匹配查找与查询相关的表。

search_tables(
    query: str,           # Search text
    database: str = "",   # Optional: filter by database
    domain: str = "",     # Optional: filter by domain
    limit: int = 5        # Max results (1-20)
)

退货： {tables: [...], total_matches, tokens_used}

______________________________________________________________________

架构工具

list_columns

获取表的列定义。

list_columns(
    table: str,           # Table name (e.g., "merchants", "DABSTEP_PAYMENTS")
    database: str = ""    # Optional: filter by database
)

退货： {table, columns: [{name, type, nullable, description}], file_path}

get_table_schema

从源JSON文件中检索完整的模式详细信息。

get_table_schema(
    table: str,                    # Table name
    database: str = "",            # Optional: filter by database
    include_samples: bool = False  # Include sample values if available
)

退货： 完整的架构包括：

• name, database, schema, description
• columns 带有类型和描述
• primary_key, foreign_keys, indexes
• related_tables, file_path

get_domain_overview

获取业务域中所有表的摘要。

get_domain_overview(
    domain: str,          # Domain name (e.g., "payments", "authentication")
    database: str = ""    # Optional: filter by database
)

退货： {domain, description, databases, tables: [{name, description}]}

______________________________________________________________________

关系工具

get_join_path

通过外键遍历找到两个表之间的连接路径。

get_join_path(
    source_table: str,    # Starting table
    target_table: str,    # Target table
    database: str = "",   # Optional: filter by database
    max_hops: int = 3     # Maximum join hops
)

退货： {source, target, found, hop_count, path: [...], sql_snippet}

get_common_relationships

根据外键列出常用的连接模式。

get_common_relationships(
    database: str = "",   # Optional: filter by database
    domain: str = "",     # Optional: filter by domain
    limit: int = 10       # Max relationships
)

退货： {relationships: [{source_table, target_table, join_sql, description}]}

______________________________________________________________________

实用工具

______________________________________________________________________

示例用法

发现可用数据

# List all databases
list_databases()
# → postgres_production (37 tables), snowflake_production (4 tables)

# List tables in a specific database
list_tables(database="snowflake_production")

# List business domains
list_domains()

搜索表格

# FTS5 search
search_fts(query="payment fraud", limit=5)

# Semantic search
search_vector(query="tables related to financial transactions", limit=3)

# Search within a specific database
search_fts(query="merchant", database="postgres_production")

获取架构详细信息

# Get schema for a table (auto-detects database)
get_table_schema(table="payments")

# Get schema from specific database
get_table_schema(table="payments", database="postgres_production")
get_table_schema(table="DABSTEP_PAYMENTS", database="snowflake_production")

# Get columns only
list_columns(table="merchants", database="postgres_production")

探索关系

# Find join path between tables
get_join_path(
    source_table="merchants",
    target_table="payments",
    database="postgres_production"
)

# Get common relationships in a domain
get_common_relationships(domain="payments")

______________________________________________________________________

建筑

┌─────────────────────────────────────────────────────────────┐
│                     MCP Client                              │
└─────────────────────────────────────────────────────────────┘
                            │
                            ▼
┌─────────────────────────────────────────────────────────────┐
│                   FastMCP Server                            │
│  ┌─────────────┐  ┌─────────────┐  ┌─────────────────────┐ │
│  │ search_fts  │  │search_vector│  │ Schema Tools        │ │
│  └──────┬──────┘  └──────┬──────┘  └──────────┬──────────┘ │
│         │                │                     │            │
│         ▼                ▼                     ▼            │
│  ┌─────────────────────────────┐    ┌─────────────────────┐│
│  │   data/index/index.db      │    │   data/map/*.json   ││
│  │  ┌───────────┐ ┌─────────┐ │    │   (source schemas)  ││
│  │  │   FTS5    │ │ Vectors │ │    └─────────────────────┘│
│  │  └───────────┘ └─────────┘ │                            │
│  └─────────────────────────────┘                            │
└─────────────────────────────────────────────────────────────┘
                            │
                            ▼ (vector search only)
                   ┌────────────────┐
                   │  OpenAI API    │
                   │  (embeddings)  │
                   └────────────────┘

______________________________________________________________________

索引内容摘要

总计： 41个表格，428个索引文档（表格+列）

______________________________________________________________________

备注

• 没有数据库连接：此MCP仅提供预索引上下文，它不连接到实际的Postgres或Snowflake数据库。
• 矢量搜索 需要 OPENAI_API_KEY 在运行时生成查询嵌入。
• FTS5搜索 在没有OpenAI的情况下，使用预构建的索引即可工作。
• 数据库参数：使用 database="postgres_production" 或 database="snowflake_production" 当表名可能重叠时，以特定数据库为目标。
• 跑 setup_db.py 在将新的JSON文件添加到 data/map/ 重建索引。