ODPS Server

Read-only MCP server for Alibaba Cloud MaxCompute with enterprise-grade security.

Features

• Read-only SQL Guard: Only SELECT queries are allowed, with automatic LIMIT injection
• Table/Column-level Access Control: Fine-grained policy via YAML allowlist
• Policy Registry: Domain-based policy management with sensitivity levels, row policies, and partition filters
• Authentication: Bearer Token and Gateway Header authentication
• Audit Logging: JSONL-format audit logs with request tracking
• Unified Response Structure: Consistent ok/data/error/metadata/audit_id format
• Join Path Catalog: Pre-approved table join relationships with business context

Quick Start

1. Install Dependencies

python -m venv .venv
.venv\Scripts\activate  # Windows
# source .venv/bin/activate  # Linux/Mac
pip install -e .

2. Configure Environment

# Copy example config
copy .env.example .env.local  # Windows
# cp .env.example .env.local  # Linux/Mac

# Edit .env.local with your ODPS credentials

Required variables:

• ALIBABA_CLOUD_ACCESS_KEY_ID
• ALIBABA_CLOUD_ACCESS_KEY_SECRET
• ODPS_ENDPOINT
• ODPS_PROJECT
• ODPS_ALLOWLIST_PATH

3. Run Server

# Streamable HTTP mode (recommended)
python -m odps_mcp_server.server

# Or using uvicorn
uvicorn odps_mcp_server.server:create_app --factory --host 0.0.0.0 --port 8000

Endpoints

Endpoint	Method	Description
/mcp	POST	Standard MCP Streamable HTTP endpoint
/	POST	Compatibility endpoint (redirects to /mcp)
/health	GET	Health check

MCP Tools

Discovery Tools

Tool	Description
list_allowed_tables	List tables accessible by the current policy
describe_table	Get table schema, columns, and partition info
get_join_paths	Get pre-approved join relationships between tables

Query Tools

Tool	Description
preview_table	Preview rows from a table with automatic partition pruning
run_select_sql	Execute a read-only SELECT query

Semantic Layer Tools (P2)

Tool	Description
search_business_terms	Search business terms by keyword
get_metric_definition	Get metric definition, expression, and dimensions
list_metrics	List available metrics, optionally by domain
get_dimensions	Get available dimensions for a metric
generate_metric_sql	Generate SQL for a metric with dimensions and filters
validate_metric_query	Validate a metric query before execution
list_domains	List all business domains

Tool Response Structure

All tools return a unified response format:

{
  "ok": true,
  "data": { ... },
  "error": null,
  "metadata": {
    "request_id": "uuid",
    "tool_name": "describe_table",
    "timestamp": "2026-05-07T15:30:00Z",
    "latency_ms": 45.2,
    "tables_used": ["saky_dw_cdm.dim_pub_product_df"],
    "columns_used": ["item_code", "standard_name"],
    "row_count": 1,
    "truncated": false
  },
  "audit_id": "uuid"
}

Authentication

Bearer Token

Include the token in the Authorization header:

curl -H "Authorization: Bearer agent-sales-001" \
  http://localhost:8000/mcp \
  -d '{"tool": "list_allowed_tables"}'

Gateway Header

When behind an API gateway, the gateway can inject identity headers:

curl -H "X-User-Id: sales_agent" \
  -H "X-Client-Id: sales_bot" \
  -H "X-Session-Id: session-xxx" \
  http://localhost:8000/mcp \
  -d '{"tool": "list_allowed_tables"}'

Token Configuration

Define tokens in config/auth.yaml:

tokens:
  agent-sales-001:
    user_id: sales_agent
    client_id: sales_bot
    allowed_tools:
      - list_allowed_tables
      - describe_table
      - preview_table
    allowed_projects:
      - saky_dw_ods
      - saky_dw_cdm
    rate_limit: "100/minute"

Policy Configuration

Policy Registry (config/policy.yaml) - Recommended

The Policy Registry provides enterprise-grade policy management with:

• Domain-based organization: Group tables by business domain (sales, supply_chain, product)
• Sensitivity levels: public, internal, confidential, restricted
• Column-level control: allowed_columns and denied_columns
• Row-level policies: Role-based row filtering
• Partition filter requirements: Enforce partition predicates
• Per-table limits: Override global limit per table

domains:
  sales:
    name: 销售域
    owner: 销售运营部
    default_project: saky_dw_ods

tables:
  saky_dw_ods.ods_sap_delivery_order_details_df:
    domain: sales
    owner: supply_chain_data_owner
    sensitivity: internal
    default_partition:
      field: pt
      strategy: max_pt
    allowed_columns:
      - delivery_ord_code
      - product_code
      - sales_amount
      - pt
    denied_columns:
      - customer_phone
      - customer_address
    row_policies:
      - role: region_manager
        predicate: region_code IN (${user.region_codes})
    max_limit: 1000
    require_partition_filter: true

default_limit: 100
max_limit: 200

security:
  deny_unapproved_join: true
  max_join_tables: 3
  query_timeout_seconds: 300

Legacy Allowlist (config/allowlist.yaml)

Simple table/column whitelist for basic use cases: saky_dw_cdm: tables: dim_pub_product_df: {} # Allow all columns saky_dw_ods: tables: ods_sap_delivery_order_details_df: allowed_columns: # Column-level restriction - delivery_ord_code - product_code - sales_amount - pt

default_limit: 100 max_limit: 200

### Join Catalog (`config/ai_catalog/joins.yaml`)

Pre-approved table join relationships:

```yaml
joins:
  - id: ods_delivery_to_dim_product
    left_table: saky_dw_ods.ods_sap_delivery_order_details_df
    right_table: saky_dw_cdm.dim_pub_product_df
    business_purpose: 为交货单明细补齐产品主数据属性
    approved: true
    confidence: high
    cardinality: N:1
    on:
      - left_field: product_code
        right_field: item_code
    example_sql: |
      SELECT o.delivery_ord_code, p.standard_name
      FROM saky_dw_ods.ods_sap_delivery_order_details_df o
      LEFT JOIN saky_dw_cdm.dim_pub_product_df p
        ON o.product_code = p.item_code
      WHERE o.pt = MAX_PT("saky_dw_ods.ods_sap_delivery_order_details_df")
      LIMIT 100;

Audit Logging

Audit logs are written to logs/audit.jsonl in JSONL format, rotated daily.

Each record contains:

{
  "audit_id": "uuid",
  "request_id": "uuid",
  "timestamp": "2026-05-07T15:30:00.123Z",
  "user_id": "sales_agent",
  "client_id": "sales_bot",
  "session_id": "session-xxx",
  "source_ip": "10.0.1.55",
  "auth_method": "bearer_token",
  "token_id": "agent-sales-001",
  "tool_name": "run_select_sql",
  "outcome": "success",
  "latency_ms": 234.5,
  "request_params": {
    "project": "saky_dw_ods",
    "sql": "SELECT ..."
  },
  "response_summary": {
    "tables_used": ["saky_dw_ods.ods_sap_delivery_order_details_df"],
    "columns_used": ["delivery_ord_code", "sales_amount"],
    "row_count": 15,
    "truncated": false
  },
  "policy_decision": "allowed",
  "deny_reason": null,
  "error_code": null,
  "error_message": null
}

Error Codes

Code	Description
AUTH_MISSING	Authentication required but not provided
AUTH_INVALID_TOKEN	Bearer token not recognized
AUTH_FORBIDDEN_ORIGIN	Origin not in allowed list
AUTH_FORBIDDEN_HOST	Host not in allowed list
AUTH_FORBIDDEN_TOOL	Tool not allowed for this token
AUTH_FORBIDDEN_PROJECT	Project not allowed for this token
TABLE_NOT_ALLOWED	Table not in allowlist
COLUMN_NOT_ALLOWED	Column not in allowed columns
SELECT_STAR_NOT_ALLOWED	SELECT * not allowed for restricted tables
SQL_PARSE_ERROR	SQL syntax error
SQL_MULTIPLE_STATEMENTS	Multiple SQL statements not allowed
SQL_READONLY_ONLY	Only SELECT queries allowed
TABLE_REFERENCE_REQUIRED	SQL must reference at least one table
LIMIT_MUST_BE_LITERAL	LIMIT must be a literal integer
LIMIT_MUST_BE_POSITIVE	LIMIT must be >= 1
RATE_LIMIT_EXCEEDED	Too many requests
INTERNAL_ERROR	Internal server error (details hidden)

Project Structure

数仓MCP/
├── src/odps_mcp_server/
│   ├── __init__.py          # Package version
│   ├── __main__.py          # Entry point
│   ├── server.py            # MCP server and ASGI app
│   ├── service.py           # Query service (business logic)
│   ├── config.py            # Configuration models
│   ├── policy.py            # Policy service (access control)
│   ├── odps_client.py       # MaxCompute client wrapper
│   ├── sql_guard.py         # SQL validation (sqlglot)
│   ├── auth.py              # Authentication module
│   ├── context.py           # Request context
│   ├── audit.py             # Audit logging
│   ├── response.py          # Unified response builder
│   ├── middleware.py         # Starlette middleware
│   ├── errors.py            # Error codes
│   └── logging_utils.py     # Logging utilities
├── config/
│   ├── allowlist.yaml       # Table access policy
│   ├── auth.yaml            # Token configuration
│   ├── security.yaml        # Origin/Host whitelist
│   └── ai_catalog/          # Table metadata and join paths
├── tests/                   # Unit tests
├── scripts/                 # Utility scripts
├── docs/                    # Documentation
├── .env.example             # Environment variables template
├── .env.local               # Local config (git-ignored)
└── pyproject.toml           # Project metadata

Manual Smoke Test

python scripts/manual_smoke_test.py list-tables --project saky_dw_ods
python scripts/manual_smoke_test.py describe-table --project saky_dw_cdm --table dim_pub_product_df
python scripts/manual_smoke_test.py run-sql --project saky_dw_ods --sql "SELECT delivery_ord_code FROM saky_dw_ods.ods_sap_delivery_order_details_df LIMIT 10"

Security Notes

• Never commit .env.local or files containing credentials
• Review config/allowlist.yaml before production deployment
• Enable authentication (MCP_REQUIRE_AUTH=true) for production
• Monitor logs/audit.jsonl for suspicious activity
• Rotate Bearer tokens periodically