connect_to_apache_iceberg
Establish a Timeplus database in Iceberg format to integrate with Iceberg data, enabling connection to specified S3 buckets and AWS regions.
Instructions
Create a Timeplus database in iceberg type to connect to Iceberg
Input Schema
| Name | Required | Description | Default |
|---|---|---|---|
| aws_account_id | Yes | ||
| aws_region | No | us-west-2 | |
| iceberg_db | Yes | ||
| is_s3_table_bucket | No | ||
| s3_bucket | Yes |
Input Schema (JSON Schema)
{
"properties": {
"aws_account_id": {
"title": "Aws Account Id",
"type": "integer"
},
"aws_region": {
"default": "us-west-2",
"title": "Aws Region",
"type": "string"
},
"iceberg_db": {
"title": "Iceberg Db",
"type": "string"
},
"is_s3_table_bucket": {
"default": false,
"title": "Is S3 Table Bucket",
"type": "boolean"
},
"s3_bucket": {
"title": "S3 Bucket",
"type": "string"
}
},
"required": [
"iceberg_db",
"aws_account_id",
"s3_bucket"
],
"title": "connect_to_apache_icebergArguments",
"type": "object"
}
Implementation Reference
- mcp_timeplus/mcp_server.py:242-264 (handler)The main handler function for the 'connect_to_apache_iceberg' tool. It creates a Timeplus database configured for Apache Iceberg integration using AWS Glue catalog and S3 storage, then lists the streams in that database.@mcp.tool() def connect_to_apache_iceberg(iceberg_db: str,aws_account_id: int, s3_bucket: str, aws_region: str="us-west-2",is_s3_table_bucket: bool=False): """Create a Timeplus database in iceberg type to connect to Iceberg""" if iceberg_db is None or aws_account_id is None or s3_bucket is None: return {"error": "iceberg_db, aws_account_id, and s3_bucket are required"} logger.info("Creating Iceberg database") warehouse=aws_account_id storage_endpoint=f"https://{s3_bucket}.s3.{aws_region}.amazonaws.com" if is_s3_table_bucket: warehouse=f"{aws_account_id}:s3tablescatalog/{s3_bucket}" storage_endpoint=f"https://s3tables.{aws_region}.amazonaws.com/{s3_bucket}" sql=f"""CREATE DATABASE {iceberg_db} SETTINGS type='iceberg',catalog_uri='https://glue.{aws_region}.amazonaws.com/iceberg',catalog_type='rest', warehouse='{warehouse}',storage_endpoint='{storage_endpoint}',use_environment_credentials=true, rest_catalog_sigv4_enabled=true,rest_catalog_signing_region='{aws_region}',rest_catalog_signing_name='glue' """ run_sql(sql) logger.info("Iceberg database created") sql=f"SHOW STREAMS FROM {iceberg_db}" return run_sql(sql)
- mcp_timeplus/__init__.py:1-23 (registration)The tool function is imported from mcp_server.py and exposed via __all__ for module-level access and likely MCP server registration.from .mcp_server import ( create_timeplus_client, list_databases, list_tables, run_sql, list_kafka_topics, explore_kafka_topic, create_kafka_stream, generate_sql, connect_to_apache_iceberg, ) __all__ = [ "list_databases", "list_tables", "run_sql", "create_timeplus_client", "list_kafka_topics", "explore_kafka_topic", "create_kafka_stream", "generate_sql", "connect_to_apache_iceberg", ]