MCP Unified DB

MCP (Model Context Protocol) server providing unified access to 27 Oracle schemas, Graylog logs, and Elasticsearch data.

Features

Tool	Description
execute_sql	Run SELECT queries against any Oracle schema
get_schemas	Load schema mapping files (full or filtered)
find_table	Search for a table across all schemas
search_schemas	Full-text search across all mappings
get_columns	List columns for a specific table
search_graylog	Query Graylog logs via Lucene syntax
list_graylog_streams	List available Graylog streams
search_elasticsearch	Execute Elasticsearch DSL queries
health_check	Validate Oracle, Graylog, and ES connections

Installation

pip install -e .

# With Elasticsearch support:
pip install -e ".[elasticsearch]"

# With dev dependencies (pytest):
pip install -e ".[dev]"

Configuration

Copy .env.example to .env and fill in your credentials:

cp .env.example .env

Required environment variables:

# Oracle (single connection shared by all 27 schemas)
ORACLE_SAE_CONN=user/password@host:1521/service_name

# Graylog
GRAYLOG_API_URL=https://logs.example.com/api
GRAYLOG_API_TOKEN=your_token

# Elasticsearch
ES_HOST=http://elastic.example.com:9200
ES_USER=admin
ES_PASS=your_password
ES_INDEX=atuacapes

Usage

As MCP Server (stdio)

python -m mcp_unified_db

Claude Code / Claude Desktop

Copy .mcp.json.example to .mcp.json and update with your credentials.

Oracle Schemas (27)

All schemas share the same connection and support unlimited cross-schema JOINs:

AVAL_PROPOSTA, CAPES, CAPES_GLOBAL, CONSPRE, CORPORATIVO, EDITAL, EDUCAPES,
EVENTOS, FINANCEIRO, INSCRICAO, PDTIC, PPG_AVALIACAO, PPG_ONLINE, PREMIO,
PREMIOCAPES, PREMIO_CAPES, PREMIO_DEFESA, PREMIO_FAMILIA, PREMIO_MINFRA,
PREMIO_VOLKSWAGEN, RABLISTENER, SAE, SCBA, SIPREC, SISCAD,
SUCUPIRA_PAINEL, WEB_MEUSDADOS

SODA (JSON Documents)

• SODA_CONSPRE - Collections: CERTIFICADOS, INSCRICOES, INSCRICOES_2, ORCID
• Access via cross-schema: SELECT * FROM SODA_CONSPRE.INSCRICOES

Recommended Workflow

1. find_table('PAGAMENTO')                          # Find which schema has the table
2. get_columns('FINANCEIRO', 'PAGAMENTO')            # List available columns
3. execute_sql(database='FINANCEIRO', sql='...')      # Execute query with confidence

Examples

Cross-schema JOIN

SELECT p.NM_PESSOA, pe.VL_PAGAMENTO
FROM SAE.PESSOA_EVENTO pe
JOIN CORPORATIVO.PESSOA p ON pe.ID_PESSOA = p.ID_PESSOA
FETCH NEXT 100 ROWS ONLY

Elasticsearch Aggregation

{
  "size": 0,
  "aggs": {
    "by_type": { "terms": { "field": "atuacoes.tipo.keyword" } }
  }
}

Graylog Lucene Query

level:error AND app:conspre AND message:*timeout*

Project Structure

mcp-unified-db/
├── pyproject.toml
├── src/mcp_unified_db/
│   ├── server.py              # FastMCP server (tools)
│   ├── instructions.py        # MCP instructions string
│   ├── config.py              # Environment configuration
│   ├── core/
│   │   ├── validators.py      # SQL validation
│   │   ├── formatters.py      # Output formatting
│   │   └── schema.py          # Schema loader/cache
│   ├── executors/
│   │   ├── oracle.py          # Oracle connection pool
│   │   ├── elasticsearch.py   # Elasticsearch DSL
│   │   └── graylog.py         # Graylog Lucene
│   └── cli/
│       ├── import_schema.py   # Import Oracle schema
│       ├── import_soda.py     # Import SODA collections
│       ├── generate_mappings.py # Batch mapping generator
│       └── report.py          # Institutional/integrated reports
├── data/schema_mappings/      # Schema mapping files (~820KB)
├── tests/
└── docs/

CLI Tools

# Import a new schema
python -m mcp_unified_db.cli.import_schema SCHEMA_NAME "Description"

# Import SODA collections
python -m mcp_unified_db.cli.import_soda

# Regenerate all mappings
python -m mcp_unified_db.cli.generate_mappings

# Generate reports
python -m mcp_unified_db.cli.report institutional
python -m mcp_unified_db.cli.report integrated

Development

pip install -e ".[dev]"
pytest tests/ -v

Testing

pytest tests/ -v
python -m py_compile src/mcp_unified_db/server.py