env_config.py•8.6 kB
"""Environment configuration for NIX MCP Server"""
import logging
import os
from typing import Dict, Optional, Tuple
import subprocess
import json
logger = logging.getLogger(__name__)
class EnvironmentConfig:
"""Manages environment-specific endpoints for NIX queries"""
# Environment endpoint mappings
ENVIRONMENTS = {
"dev": {
"nodeos": "http://producer-1-a-sin-dev.node.c-sin-g-atoms-b1fs-dev.int.b1fs.net:8888",
"rodeos": "http://rodeos-wasm-ql-blockchain-atoms-b1fs-dev.service.c-sin-g-atoms-b1fs-dev.int.b1fs.net:8880",
"signature_provider": "EOS69j9W5nqhwzv6VZxmoTjehE8LKURszRwEccmmwxMsBE3WjEyaa=KEY:5JCiYtQ9qsjifLw8XGQCWsg2sd33mfWJ7aR8Hy8tSxf2A4suvrc"
},
"uat": {
"nodeos": "http://producer-1-a-sin-uat.node.c-sin-g-atoms-b1fs-dev.int.b1fs.net:8888",
"rodeos": "http://rodeos-wasm-ql-blockchain-atoms-b1fs-uat.service.c-sin-g-atoms-b1fs-dev.int.b1fs.net:8880",
"signature_provider": "EOS69j9W5nqhwzv6VZxmoTjehE8LKURszRwEccmmwxMsBE3WjEyaa=KEY:5JCiYtQ9qsjifLw8XGQCWsg2sd33mfWJ7aR8Hy8tSxf2A4suvrc"
},
"cdev": {
"nodeos": "http://producer-1-a-sin-custody.node.c-sin-g-atoms-b1fs-dev.int.b1fs.net:8888",
"rodeos": "http://rodeos-wasm-ql-blockchain-atoms-b1fs-custody.service.c-sin-g-atoms-b1fs-dev.int.b1fs.net:8880",
"signature_provider": "EOS7gBMMoGKRGAEnVHcQ4YbefiksmBFPN1hZXSWuhUxcu85W2orQt=KEY:5KEv2EH8n45zS4ggfv4FeG3wVN5E18X8HafmSuQAJURVuK7QmKw"
},
"perf": {
"nodeos": "http://producer-1-a-sin-perf.node.c-sin-g-atoms-b1fs-perf.int.b1fs.net:8888",
"rodeos": "http://rodeos-wasm-ql-blockchain-atoms-b1fs-perf.service.c-sin-g-atoms-b1fs-perf.int.b1fs.net:8880",
"signature_provider": None
},
"perf2": {
"nodeos": "http://producer-1-a-sin-perf2.node.c-sin-g-atoms-b1fs-perf.int.b1fs.net:8888",
"rodeos": "http://rodeos-wasm-ql-blockchain-atoms-b1fs-perf2.service.c-sin-g-atoms-b1fs-perf.int.b1fs.net:8880",
"signature_provider": None
},
"simnext": {
"nodeos": "http://producer-1-a-sin-simnext.node.c-sin-g-atoms-b1fs-pub.int.b1fs.net:8888",
"rodeos": "http://rodeos-wasm-ql-blockchain-atoms-b1fs-simnext.service.c-sin-g-atoms-b1fs-pub.int.b1fs.net:8880",
"signature_provider": "EOS69j9W5nqhwzv6VZxmoTjehE8LKURszRwEccmmwxMsBE3WjEyaa=KEY:5JCiYtQ9qsjifLw8XGQCWsg2sd33mfWJ7aR8Hy8tSxf2A4suvrc"
},
"prod": {
"nodeos": "http://producer-1-a-sin-prod.node.c-sin-g-atoms-b1fs-prod.int.b1fs.net:8888",
"rodeos": "http://rodeos-wasm-ql-blockchain-atoms-b1fs-prod.service.c-sin-g-atoms-b1fs-prod.int.b1fs.net:8880",
"signature_provider": None # Production signature provider should not be hardcoded
},
"local": {
"nodeos": "http://localhost:8888",
"rodeos": "http://localhost:8880",
"signature_provider": "EOS6MRyAjQq8ud7hVNYcfnVPJqcVpscN5So8BhtHuGYqET5GDW5CV=KEY:5KQwrPbwdL6PhXujxW37FSSQZ1JiwsST4cqQzDeyXtP79zkvFD3"
},
"snapshot": {
"nodeos": "http://localhost:8888",
"rodeos": "http://localhost:8880",
"signature_provider": "EOS6MRyAjQq8ud7hVNYcfnVPJqcVpscN5So8BhtHuGYqET5GDW5CV=KEY:5KQwrPbwdL6PhXujxW37FSSQZ1JiwsST4cqQzDeyXtP79zkvFD3"
}
}
def __init__(self, default_env: str = "cdev"):
"""
Initialize environment configuration
Args:
default_env: Default environment to use
"""
self.default_env = default_env
self._validate_environment(default_env)
def _validate_environment(self, env: str) -> None:
"""
Validate that environment exists
Args:
env: Environment name to validate
Raises:
ValueError: If environment is not valid
"""
if env not in self.ENVIRONMENTS:
valid_envs = ", ".join(self.ENVIRONMENTS.keys())
raise ValueError(f"Invalid environment: {env}. Valid options: {valid_envs}")
def get_endpoints(self, environment: Optional[str] = None) -> Tuple[str, str]:
"""
Get Nodeos and Rodeos endpoints for an environment
Args:
environment: Environment name (defaults to default_env)
Returns:
Tuple of (nodeos_api, rodeos_api)
"""
env = environment or self.default_env
self._validate_environment(env)
# Check if environment variables override the defaults
env_upper = env.upper()
nodeos_env_var = f"NODEOS_API_{env_upper}"
rodeos_env_var = f"RODEOS_API_{env_upper}"
nodeos_api = os.getenv(nodeos_env_var) or self.ENVIRONMENTS[env]["nodeos"]
rodeos_api = os.getenv(rodeos_env_var) or self.ENVIRONMENTS[env]["rodeos"]
# For environments that support leader refresh, try to get the current leader
if env in ["dev", "cdev", "uat", "simnext", "prod"]:
try:
leader_endpoint = self._refresh_leader(nodeos_api)
if leader_endpoint:
nodeos_api = leader_endpoint
logger.info(f"Using leader endpoint for {env}: {nodeos_api}")
except Exception as e:
logger.debug(f"Could not refresh leader for {env}: {e}")
return nodeos_api, rodeos_api
def _refresh_leader(self, nodeos_api: str) -> Optional[str]:
"""
Try to get the current leader node endpoint
Args:
nodeos_api: Current nodeos endpoint
Returns:
Leader endpoint if available, None otherwise
"""
try:
# Use curl to get leader info (more reliable than Python requests for internal endpoints)
cmd = ["curl", "--silent", "-S", f"{nodeos_api}/v1/producer_ha/get_info"]
result = subprocess.run(cmd, capture_output=True, text=True, timeout=5)
if result.returncode == 0:
data = json.loads(result.stdout)
leader_id = data.get("leader_id")
if leader_id:
# Find the leader's address
for peer in data.get("peers", []):
if peer.get("id") == leader_id:
address = peer.get("address", "")
# Replace port 8988 with 8888 for API endpoint
address = address.replace(":8988", ":8888")
if address:
return f"http://{address}"
return None
except Exception as e:
logger.debug(f"Error refreshing leader: {e}")
return None
def get_signature_provider(self, environment: Optional[str] = None) -> Optional[str]:
"""
Get signature provider for an environment
Args:
environment: Environment name (defaults to default_env)
Returns:
Signature provider string or None
"""
env = environment or self.default_env
self._validate_environment(env)
# Check environment variable first
env_upper = env.upper()
sig_env_var = f"SIGNATURE_PROVIDER_{env_upper}"
return os.getenv(sig_env_var) or self.ENVIRONMENTS[env].get("signature_provider")
@classmethod
def list_environments(cls) -> Dict[str, Dict[str, str]]:
"""
List all available environments and their endpoints
Returns:
Dictionary of all environments
"""
return cls.ENVIRONMENTS.copy()
@classmethod
def get_environment_info(cls, environment: str) -> Dict[str, str]:
"""
Get detailed information about a specific environment
Args:
environment: Environment name
Returns:
Dictionary with environment details
"""
if environment not in cls.ENVIRONMENTS:
raise ValueError(f"Unknown environment: {environment}")
return {
"name": environment,
"nodeos": cls.ENVIRONMENTS[environment]["nodeos"],
"rodeos": cls.ENVIRONMENTS[environment]["rodeos"],
"has_signature": cls.ENVIRONMENTS[environment].get("signature_provider") is not None
}