diff --git a/pkg/server/builder.go b/pkg/server/builder.go index 548b8ce..6ddc21d 100644 --- a/pkg/server/builder.go +++ b/pkg/server/builder.go @@ -20,6 +20,7 @@ import ( clickhouseplugin "github.com/ethpandaops/mcp/plugins/clickhouse" doraplugin "github.com/ethpandaops/mcp/plugins/dora" + labplugin "github.com/ethpandaops/mcp/plugins/lab" lokiplugin "github.com/ethpandaops/mcp/plugins/loki" prometheusplugin "github.com/ethpandaops/mcp/plugins/prometheus" ) @@ -201,6 +202,7 @@ func (b *Builder) buildPluginRegistry() (*plugin.Registry, error) { // Register all compiled-in plugins. reg.Add(clickhouseplugin.New()) reg.Add(doraplugin.New()) + reg.Add(labplugin.New()) reg.Add(lokiplugin.New()) reg.Add(prometheusplugin.New()) diff --git a/plugins/lab/config.go b/plugins/lab/config.go new file mode 100644 index 0000000..4d599ea --- /dev/null +++ b/plugins/lab/config.go @@ -0,0 +1,47 @@ +package lab + +// Config holds the Lab plugin configuration. +// Lab is enabled by default since it's a public service and +// requires no credentials. +type Config struct { + // Enabled controls whether the Lab plugin is active. + // Defaults to true. + Enabled *bool `yaml:"enabled,omitempty"` + + // RoutesURL is the URL to fetch the lab routes.json from. + // Defaults to https://raw.githubusercontent.com/ethpandaops/lab/main/routes.json + RoutesURL string `yaml:"routes_url,omitempty"` + + // SkillURL is the URL to fetch the lab SKILL.md from. + // Defaults to https://raw.githubusercontent.com/ethpandaops/lab/main/SKILL.md + SkillURL string `yaml:"skill_url,omitempty"` + + // Networks allows specifying custom Lab URLs for networks. + // If not specified, default URLs will be used for known networks. + Networks map[string]string `yaml:"networks,omitempty"` +} + +// IsEnabled returns true if the plugin is enabled (default: true). +func (c *Config) IsEnabled() bool { + if c.Enabled == nil { + return true + } + + return *c.Enabled +} + +// GetRoutesURL returns the routes.json URL with default. +func (c *Config) GetRoutesURL() string { + if c.RoutesURL == "" { + return "https://raw.githubusercontent.com/ethpandaops/lab/main/routes.json" + } + return c.RoutesURL +} + +// GetSkillURL returns the SKILL.md URL with default. +func (c *Config) GetSkillURL() string { + if c.SkillURL == "" { + return "https://raw.githubusercontent.com/ethpandaops/lab/main/SKILL.md" + } + return c.SkillURL +} diff --git a/plugins/lab/examples.go b/plugins/lab/examples.go new file mode 100644 index 0000000..db40041 --- /dev/null +++ b/plugins/lab/examples.go @@ -0,0 +1,29 @@ +package lab + +import ( + _ "embed" + "fmt" + "strings" + + "github.com/ethpandaops/mcp/pkg/types" + "gopkg.in/yaml.v3" +) + +//go:embed examples.yaml +var examplesYAML []byte + +var queryExamples map[string]types.ExampleCategory + +func init() { + if err := yaml.Unmarshal(examplesYAML, &queryExamples); err != nil { + panic(fmt.Sprintf("failed to parse lab examples.yaml: %v", err)) + } + + for key, category := range queryExamples { + for i := range category.Examples { + category.Examples[i].Query = strings.TrimSpace(category.Examples[i].Query) + } + + queryExamples[key] = category + } +} diff --git a/plugins/lab/examples.yaml b/plugins/lab/examples.yaml new file mode 100644 index 0000000..0b6331c --- /dev/null +++ b/plugins/lab/examples.yaml @@ -0,0 +1,187 @@ +lab_network_discovery: + name: Network Discovery + description: Discover Lab-enabled networks and available routes + examples: + - name: List Lab networks + description: Get all networks with Lab explorers + query: | + from ethpandaops import lab + + # List available networks with Lab explorers + networks = lab.list_networks() + print(f"Available networks: {[n['name'] for n in networks]}") + + # Get base URL for a specific network + if networks: + base_url = lab.get_base_url(networks[0]['name']) + print(f"Base URL: {base_url}") + + - name: Get all routes + description: Fetch and display all Lab routes from routes.json + query: | + from ethpandaops import lab + + # Get all routes organized by category + routes = lab.get_routes() + + for category, route_list in routes.items(): + print(f"\n{category}:") + for route in route_list: + print(f" - {route['id']}: {route.get('description', 'No description')}") + + - name: Get routes by category + description: Filter routes to a specific category + query: | + from ethpandaops import lab + + # Get only Ethereum routes + eth_routes = lab.get_routes_by_category("ethereum") + print(f"Ethereum routes ({len(eth_routes)}):") + for route in eth_routes: + print(f" - {route['id']}: {route.get('path', 'N/A')}") + +lab_deep_links: + name: Deep Links + description: Generate Lab explorer links for various Ethereum entities + examples: + - name: Generate slot and epoch links + description: Create deep links to slots and epochs + query: | + from ethpandaops import lab + + network = "mainnet" + slot = 1000000 + epoch = 31250 + + # Slot link + slot_link = lab.link_slot(network, slot) + print(f"Slot {slot}: {slot_link}") + + # Epoch link + epoch_link = lab.link_epoch(network, epoch) + print(f"Epoch {epoch}: {epoch_link}") + + - name: Generate validator and address links + description: Create deep links to validators and addresses + query: | + from ethpandaops import lab + + network = "sepolia" + + # Validator link + validator_link = lab.link_validator(network, "12345") + print(f"Validator: {validator_link}") + + # Address link + address_link = lab.link_address(network, "0x742d35Cc6634C0532925a3b844Bc9e7595f1c1") + print(f"Address: {address_link}") + + - name: Generate block and transaction links + description: Create deep links to blocks and transactions + query: | + from ethpandaops import lab + + network = "holesky" + + # Block link + block_link = lab.link_block(network, "1000000") + print(f"Block: {block_link}") + + # Transaction link + tx_link = lab.link_transaction(network, "0xabc123...") + print(f"Transaction: {tx_link}") + + - name: Generate blob and fork links + description: Create deep links to blobs and fork information + query: | + from ethpandaops import lab + + network = "mainnet" + + # Blob link + blob_link = lab.link_blob(network, "0x1234...") + print(f"Blob: {blob_link}") + + # Fork link + fork_link = lab.link_fork(network, "cancun") + print(f"Fork: {fork_link}") + +lab_custom_urls: + name: Custom URL Building + description: Build custom URLs using route patterns and parameters + examples: + - name: Build URLs with parameters + description: Use build_url for flexible URL generation + query: | + from ethpandaops import lab + + # Build a URL for a specific route + url = lab.build_url("mainnet", "slot", {"slot": 1000000}) + print(f"Custom URL: {url}") + + # Get route metadata first + route = lab.get_route("ethereum/slots") + if route: + print(f"Route path: {route.get('path')}") + print(f"Parameters: {route.get('parameters', [])}") + + - name: Multi-network link generation + description: Generate links across multiple networks + query: | + from ethpandaops import lab + + networks = ["mainnet", "sepolia", "holesky", "hoodi"] + slot = 1000000 + + print(f"Slot {slot} across networks:") + for network in networks: + try: + link = lab.link_slot(network, slot) + print(f" {network}: {link}") + except ValueError as e: + print(f" {network}: Not available ({e})") + +lab_combined_workflows: + name: Combined Workflows + description: Combine Lab with other data sources for comprehensive analysis + examples: + - name: Network analysis with Lab links + description: Combine Lab with ClickHouse data + query: | + from ethpandaops import lab, clickhouse + + network = "holesky" + + # Get recent blocks from ClickHouse + df = clickhouse.query("xatu", f''' + SELECT + slot, + block_root, + proposer_index + FROM beacon_api_eth_v1_events_block + WHERE meta_network_name = '{network}' + ORDER BY slot DESC + LIMIT 5 + ''') + + print(f"Recent blocks on {network}:") + for _, row in df.iterrows(): + link = lab.link_slot(network, row['slot']) + print(f" Slot {row['slot']}: proposer {row['proposer_index']}") + print(f" View: {link}") + + - name: Cross-explorer comparison + description: Compare Lab and Dora links for the same entity + query: | + from ethpandaops import lab, dora + + network = "holesky" + slot = 1000000 + + # Get links from both explorers + lab_link = lab.link_slot(network, slot) + dora_link = dora.link_slot(network, str(slot)) + + print(f"Slot {slot} explorer links:") + print(f" Lab: {lab_link}") + print(f" Dora: {dora_link}") diff --git a/plugins/lab/plugin.go b/plugins/lab/plugin.go new file mode 100644 index 0000000..da2700f --- /dev/null +++ b/plugins/lab/plugin.go @@ -0,0 +1,208 @@ +package lab + +import ( + "context" + "encoding/json" + "fmt" + + "github.com/sirupsen/logrus" + "gopkg.in/yaml.v3" + + "github.com/ethpandaops/mcp/pkg/plugin" + "github.com/ethpandaops/mcp/pkg/resource" + "github.com/ethpandaops/mcp/pkg/types" +) + +// defaultLabURLs provides fallback Lab URLs for known networks. +var defaultLabURLs = map[string]string{ + "mainnet": "https://lab.ethpandaops.io", + "sepolia": "https://lab.ethpandaops.io/sepolia", + "holesky": "https://lab.ethpandaops.io/holesky", + "hoodi": "https://lab.ethpandaops.io/hoodi", +} + +// Plugin implements the plugin.Plugin interface for Lab explorer. +type Plugin struct { + cfg Config + cartographoorClient resource.CartographoorClient + log logrus.FieldLogger +} + +// New creates a new Lab plugin. +func New() *Plugin { + return &Plugin{} +} + +func (p *Plugin) Name() string { return "lab" } + +// DefaultEnabled implements plugin.DefaultEnabled. +// Lab is enabled by default since it requires no configuration. +func (p *Plugin) DefaultEnabled() bool { return true } + +func (p *Plugin) Init(rawConfig []byte) error { + if len(rawConfig) == 0 { + // No config provided, use defaults (enabled = true). + return nil + } + + return yaml.Unmarshal(rawConfig, &p.cfg) +} + +func (p *Plugin) ApplyDefaults() { + // Defaults are handled by Config methods. +} + +func (p *Plugin) Validate() error { + // No validation needed - config is minimal. + return nil +} + +// SandboxEnv returns environment variables for the sandbox. +// Returns ETHPANDAOPS_LAB_NETWORKS with network->URL mapping. +func (p *Plugin) SandboxEnv() (map[string]string, error) { + if !p.cfg.IsEnabled() { + return nil, nil + } + + // Build network -> Lab URL mapping from config or defaults. + labNetworks := make(map[string]string) + + // First, add any custom networks from config + if p.cfg.Networks != nil { + for name, url := range p.cfg.Networks { + labNetworks[name] = url + } + } + + // Then, add known networks from cartographoor if available + if p.cartographoorClient != nil { + networks := p.cartographoorClient.GetActiveNetworks() + for name := range networks { + if url, ok := defaultLabURLs[name]; ok { + if _, alreadySet := labNetworks[name]; !alreadySet { + labNetworks[name] = url + } + } + } + } + + // Ensure default networks are always available + for name, url := range defaultLabURLs { + if _, ok := labNetworks[name]; !ok { + labNetworks[name] = url + } + } + + networksJSON, err := json.Marshal(labNetworks) + if err != nil { + return nil, fmt.Errorf("marshaling lab networks: %w", err) + } + + // Also pass routes and skill URLs + env := map[string]string{ + "ETHPANDAOPS_LAB_NETWORKS": string(networksJSON), + "ETHPANDAOPS_LAB_ROUTES_URL": p.cfg.GetRoutesURL(), + "ETHPANDAOPS_LAB_SKILL_URL": p.cfg.GetSkillURL(), + } + + return env, nil +} + +// DatasourceInfo returns empty since networks are the datasources, +// and those come from cartographoor. +func (p *Plugin) DatasourceInfo() []types.DatasourceInfo { + return nil +} + +func (p *Plugin) Examples() map[string]types.ExampleCategory { + if !p.cfg.IsEnabled() { + return nil + } + + result := make(map[string]types.ExampleCategory, len(queryExamples)) + for k, v := range queryExamples { + result[k] = v + } + + return result +} + +func (p *Plugin) PythonAPIDocs() map[string]types.ModuleDoc { + if !p.cfg.IsEnabled() { + return nil + } + + return map[string]types.ModuleDoc{ + "lab": { + Description: "Query Lab explorer and generate deep links using routes.json and SKILL.md patterns", + Functions: map[string]types.FunctionDoc{ + "list_networks": {Signature: "list_networks() -> list[dict]", Description: "List networks with Lab explorers"}, + "get_base_url": {Signature: "get_base_url(network) -> str", Description: "Get Lab base URL for a network"}, + "get_routes": {Signature: "get_routes() -> dict", Description: "Get all Lab routes from routes.json"}, + "get_routes_by_category": {Signature: "get_routes_by_category(category) -> list[dict]", Description: "Get routes filtered by category"}, + "get_route": {Signature: "get_route(route_id) -> dict", Description: "Get route metadata by ID"}, + "link_slot": {Signature: "link_slot(network, slot) -> str", Description: "Deep link to slot"}, + "link_epoch": {Signature: "link_epoch(network, epoch) -> str", Description: "Deep link to epoch"}, + "link_validator": {Signature: "link_validator(network, index_or_pubkey) -> str", Description: "Deep link to validator"}, + "link_block": {Signature: "link_block(network, number_or_hash) -> str", Description: "Deep link to block"}, + "link_address": {Signature: "link_address(network, address) -> str", Description: "Deep link to address"}, + "link_transaction": {Signature: "link_transaction(network, tx_hash) -> str", Description: "Deep link to transaction"}, + "link_blob": {Signature: "link_blob(network, blob_id) -> str", Description: "Deep link to blob"}, + "link_fork": {Signature: "link_fork(network, fork_name) -> str", Description: "Deep link to fork"}, + "build_url": {Signature: "build_url(network, route_id, params) -> str", Description: "Build URL for route with parameters"}, + }, + }, + } +} + +func (p *Plugin) GettingStartedSnippet() string { + if !p.cfg.IsEnabled() { + return "" + } + + return `## Lab Explorer + +Query the Lab explorer for Ethereum network data and generate deep links. +Lab uses routes.json for URL patterns and SKILL.md for deep linking. + +` + "```python" + ` +from ethpandaops import lab + +# List networks with Lab explorers +networks = lab.list_networks() + +# Get available routes +routes = lab.get_routes() +print(f"Available categories: {list(routes.keys())}") + +# Generate deep links +link = lab.link_slot("mainnet", 1000000) +print(f"View slot: {link}") + +# Build custom URLs +url = lab.build_url("sepolia", "epoch", {"epoch": 5000}) +` + "```" + ` +` +} + +// SetCartographoorClient implements plugin.CartographoorAware. +// This is called by the builder to inject the cartographoor client. +func (p *Plugin) SetCartographoorClient(client any) { + if c, ok := client.(resource.CartographoorClient); ok { + p.cartographoorClient = c + } +} + +// SetLogger sets the logger for the plugin. +func (p *Plugin) SetLogger(log logrus.FieldLogger) { + p.log = log.WithField("plugin", "lab") +} + +// RegisterResources is a no-op since Lab uses networks:// resources. +func (p *Plugin) RegisterResources(_ logrus.FieldLogger, _ plugin.ResourceRegistry) error { + return nil +} + +func (p *Plugin) Start(_ context.Context) error { return nil } + +func (p *Plugin) Stop(_ context.Context) error { return nil } diff --git a/plugins/lab/python/lab.py b/plugins/lab/python/lab.py new file mode 100644 index 0000000..94b602f --- /dev/null +++ b/plugins/lab/python/lab.py @@ -0,0 +1,306 @@ +"""Lab explorer access. + +Query the Lab explorer and generate deep links using routes.json patterns. +Network URLs are discovered from cartographoor via environment variables. + +Example: + from ethpandaops import lab + + networks = lab.list_networks() + routes = lab.get_routes() + link = lab.link_slot("mainnet", 1000000) +""" + +import json +import os +import re +from typing import Any + +import httpx + +_NETWORKS: dict[str, str] | None = None +_ROUTES: dict[str, Any] | None = None +_SKILL_PATTERNS: dict[str, Any] | None = None +_TIMEOUT = httpx.Timeout(connect=5.0, read=30.0, write=10.0, pool=5.0) + +# Default Lab URLs for known networks (fallback if cartographoor doesn't have Lab) +_DEFAULT_LAB_URLS = { + "mainnet": "https://lab.ethpandaops.io", + "sepolia": "https://lab.ethpandaops.io/sepolia", + "holesky": "https://lab.ethpandaops.io/holesky", + "hoodi": "https://lab.ethpandaops.io/hoodi", +} + + +def _load_networks() -> dict[str, str]: + """Load network -> Lab URL mapping from environment.""" + global _NETWORKS + if _NETWORKS is not None: + return _NETWORKS + + raw = os.environ.get("ETHPANDAOPS_LAB_NETWORKS", "") + if raw: + try: + _NETWORKS = json.loads(raw) + except json.JSONDecodeError as e: + raise ValueError(f"Invalid ETHPANDAOPS_LAB_NETWORKS JSON: {e}") from e + else: + _NETWORKS = {} + + # Merge with defaults for any missing networks + for network, url in _DEFAULT_LAB_URLS.items(): + if network not in _NETWORKS: + _NETWORKS[network] = url + + return _NETWORKS + + +def _get_url(network: str) -> str: + """Get Lab base URL for a network.""" + networks = _load_networks() + if network not in networks: + # Try to construct from default pattern + if network in _DEFAULT_LAB_URLS: + return _DEFAULT_LAB_URLS[network] + raise ValueError(f"Unknown network '{network}'. Available: {list(networks.keys())}") + return networks[network] + + +def _fetch_routes() -> dict[str, Any]: + """Fetch routes.json from the configured URL.""" + global _ROUTES + if _ROUTES is not None: + return _ROUTES + + routes_url = os.environ.get( + "ETHPANDAOPS_LAB_ROUTES_URL", + "https://raw.githubusercontent.com/ethpandaops/lab/main/routes.json" + ) + + try: + with httpx.Client(timeout=_TIMEOUT) as client: + resp = client.get(routes_url) + resp.raise_for_status() + _ROUTES = resp.json() + except Exception as e: + # Return default routes structure if fetch fails + _ROUTES = _get_default_routes() + + return _ROUTES + + +def _get_default_routes() -> dict[str, Any]: + """Get default routes structure.""" + return { + "ethereum": [ + { + "id": "ethereum/slots", + "path": "/ethereum/slots/{slot}", + "description": "Ethereum slot details", + "parameters": [{"name": "slot", "type": "integer", "required": True}], + }, + { + "id": "ethereum/epochs", + "path": "/ethereum/epochs/{epoch}", + "description": "Ethereum epoch details", + "parameters": [{"name": "epoch", "type": "integer", "required": True}], + }, + { + "id": "ethereum/validators", + "path": "/ethereum/validators/{validator}", + "description": "Validator details by index or pubkey", + "parameters": [{"name": "validator", "type": "string", "required": True}], + }, + { + "id": "ethereum/blocks", + "path": "/ethereum/blocks/{block}", + "description": "Execution layer block details", + "parameters": [{"name": "block", "type": "string", "required": True}], + }, + { + "id": "ethereum/transactions", + "path": "/ethereum/transactions/{tx_hash}", + "description": "Transaction details", + "parameters": [{"name": "tx_hash", "type": "string", "required": True}], + }, + { + "id": "ethereum/addresses", + "path": "/ethereum/addresses/{address}", + "description": "Address details", + "parameters": [{"name": "address", "type": "string", "required": True}], + }, + { + "id": "ethereum/blobs", + "path": "/ethereum/blobs/{blob_id}", + "description": "Blob details", + "parameters": [{"name": "blob_id", "type": "string", "required": True}], + }, + { + "id": "ethereum/forks", + "path": "/ethereum/forks/{fork_name}", + "description": "Fork information", + "parameters": [{"name": "fork_name", "type": "string", "required": True}], + }, + ] + } + + +def _get_route_by_id(route_id: str) -> dict[str, Any] | None: + """Get route metadata by ID.""" + routes = _fetch_routes() + for category, route_list in routes.items(): + for route in route_list: + if route.get("id") == route_id: + return route + return None + + +def _build_path(path_template: str, params: dict[str, Any]) -> str: + """Build a path from a template and parameters.""" + result = path_template + for key, value in params.items(): + result = result.replace(f"{{{key}}}", str(value)) + return result + + +# Network discovery + + +def list_networks() -> list[dict[str, str]]: + """List networks with Lab explorers.""" + return [{"name": n, "lab_url": u} for n, u in sorted(_load_networks().items())] + + +def get_base_url(network: str) -> str: + """Get Lab base URL for a network.""" + return _get_url(network) + + +# Routes API + + +def get_routes() -> dict[str, Any]: + """Get all Lab routes from routes.json.""" + return _fetch_routes() + + +def get_routes_by_category(category: str) -> list[dict[str, Any]]: + """Get routes filtered by category.""" + routes = _fetch_routes() + return routes.get(category, []) + + +def get_route(route_id: str) -> dict[str, Any] | None: + """Get route metadata by ID.""" + return _get_route_by_id(route_id) + + +# Deep links + + +def link_slot(network: str, slot: int) -> str: + """Generate link to slot page.""" + base = _get_url(network) + route = _get_route_by_id("ethereum/slots") + if route: + path = _build_path(route.get("path", "/ethereum/slots/{slot}"), {"slot": slot}) + return f"{base}{path}" + return f"{base}/ethereum/slots/{slot}" + + +def link_epoch(network: str, epoch: int) -> str: + """Generate link to epoch page.""" + base = _get_url(network) + route = _get_route_by_id("ethereum/epochs") + if route: + path = _build_path(route.get("path", "/ethereum/epochs/{epoch}"), {"epoch": epoch}) + return f"{base}{path}" + return f"{base}/ethereum/epochs/{epoch}" + + +def link_validator(network: str, index_or_pubkey: str) -> str: + """Generate link to validator page.""" + base = _get_url(network) + route = _get_route_by_id("ethereum/validators") + if route: + path = _build_path(route.get("path", "/ethereum/validators/{validator}"), {"validator": index_or_pubkey}) + return f"{base}{path}" + return f"{base}/ethereum/validators/{index_or_pubkey}" + + +def link_block(network: str, number_or_hash: str) -> str: + """Generate link to execution block page.""" + base = _get_url(network) + route = _get_route_by_id("ethereum/blocks") + if route: + path = _build_path(route.get("path", "/ethereum/blocks/{block}"), {"block": number_or_hash}) + return f"{base}{path}" + return f"{base}/ethereum/blocks/{number_or_hash}" + + +def link_address(network: str, address: str) -> str: + """Generate link to address page.""" + base = _get_url(network) + route = _get_route_by_id("ethereum/addresses") + if route: + path = _build_path(route.get("path", "/ethereum/addresses/{address}"), {"address": address}) + return f"{base}{path}" + return f"{base}/ethereum/addresses/{address}" + + +def link_transaction(network: str, tx_hash: str) -> str: + """Generate link to transaction page.""" + base = _get_url(network) + route = _get_route_by_id("ethereum/transactions") + if route: + path = _build_path(route.get("path", "/ethereum/transactions/{tx_hash}"), {"tx_hash": tx_hash}) + return f"{base}{path}" + return f"{base}/ethereum/transactions/{tx_hash}" + + +def link_blob(network: str, blob_id: str) -> str: + """Generate link to blob page.""" + base = _get_url(network) + route = _get_route_by_id("ethereum/blobs") + if route: + path = _build_path(route.get("path", "/ethereum/blobs/{blob_id}"), {"blob_id": blob_id}) + return f"{base}{path}" + return f"{base}/ethereum/blobs/{blob_id}" + + +def link_fork(network: str, fork_name: str) -> str: + """Generate link to fork information page.""" + base = _get_url(network) + route = _get_route_by_id("ethereum/forks") + if route: + path = _build_path(route.get("path", "/ethereum/forks/{fork_name}"), {"fork_name": fork_name}) + return f"{base}{path}" + return f"{base}/ethereum/forks/{fork_name}" + + +def build_url(network: str, route_id: str, params: dict[str, Any]) -> str: + """Build URL for a specific route with parameters. + + Args: + network: Network name (e.g., "mainnet", "sepolia") + route_id: Route identifier (e.g., "ethereum/slots") + params: Dictionary of parameters to substitute in the route path + + Returns: + Complete URL for the route + + Raises: + ValueError: If network or route is not found + """ + base = _get_url(network) + route = _get_route_by_id(route_id) + + if not route: + raise ValueError(f"Unknown route '{route_id}'") + + path_template = route.get("path", "") + if not path_template: + raise ValueError(f"Route '{route_id}' has no path defined") + + path = _build_path(path_template, params) + return f"{base}{path}" diff --git a/sandbox/ethpandaops/ethpandaops/__init__.py b/sandbox/ethpandaops/ethpandaops/__init__.py index 8c95089..00fcbb5 100644 --- a/sandbox/ethpandaops/ethpandaops/__init__.py +++ b/sandbox/ethpandaops/ethpandaops/__init__.py @@ -5,12 +5,14 @@ - Prometheus: Infrastructure metrics - Loki: Log data - Storage: S3-compatible file storage for outputs +- Lab: Ethereum explorer deep links using routes.json +- Dora: Beacon chain explorer access Use list_datasources() on each module to discover available datasources or check the datasources://list MCP resource. Example usage: - from ethpandaops import clickhouse, prometheus, loki, storage + from ethpandaops import clickhouse, prometheus, loki, storage, lab, dora # List available ClickHouse clusters clusters = clickhouse.list_datasources() @@ -22,6 +24,9 @@ # Query Prometheus using instance name result = prometheus.query("ethpandaops", "up") + # Generate Lab deep links + link = lab.link_slot("mainnet", 1000000) + # Upload output file url = storage.upload("/workspace/chart.png") """ @@ -29,14 +34,14 @@ from . import storage # Plugin modules are assembled at Docker build time -# and can be imported as: from ethpandaops import clickhouse, prometheus, loki +# and can be imported as: from ethpandaops import clickhouse, prometheus, loki, dora, lab __all__ = ["storage"] __version__ = "0.1.0" def __getattr__(name): - """Lazy import for plugin modules (clickhouse, prometheus, loki, dora).""" - if name in ("clickhouse", "prometheus", "loki", "dora"): + """Lazy import for plugin modules (clickhouse, prometheus, loki, dora, lab).""" + if name in ("clickhouse", "prometheus", "loki", "dora", "lab"): import importlib mod = importlib.import_module(f".{name}", __name__) diff --git a/sandbox/ethpandaops/ethpandaops/lab.py b/sandbox/ethpandaops/ethpandaops/lab.py new file mode 100644 index 0000000..94b602f --- /dev/null +++ b/sandbox/ethpandaops/ethpandaops/lab.py @@ -0,0 +1,306 @@ +"""Lab explorer access. + +Query the Lab explorer and generate deep links using routes.json patterns. +Network URLs are discovered from cartographoor via environment variables. + +Example: + from ethpandaops import lab + + networks = lab.list_networks() + routes = lab.get_routes() + link = lab.link_slot("mainnet", 1000000) +""" + +import json +import os +import re +from typing import Any + +import httpx + +_NETWORKS: dict[str, str] | None = None +_ROUTES: dict[str, Any] | None = None +_SKILL_PATTERNS: dict[str, Any] | None = None +_TIMEOUT = httpx.Timeout(connect=5.0, read=30.0, write=10.0, pool=5.0) + +# Default Lab URLs for known networks (fallback if cartographoor doesn't have Lab) +_DEFAULT_LAB_URLS = { + "mainnet": "https://lab.ethpandaops.io", + "sepolia": "https://lab.ethpandaops.io/sepolia", + "holesky": "https://lab.ethpandaops.io/holesky", + "hoodi": "https://lab.ethpandaops.io/hoodi", +} + + +def _load_networks() -> dict[str, str]: + """Load network -> Lab URL mapping from environment.""" + global _NETWORKS + if _NETWORKS is not None: + return _NETWORKS + + raw = os.environ.get("ETHPANDAOPS_LAB_NETWORKS", "") + if raw: + try: + _NETWORKS = json.loads(raw) + except json.JSONDecodeError as e: + raise ValueError(f"Invalid ETHPANDAOPS_LAB_NETWORKS JSON: {e}") from e + else: + _NETWORKS = {} + + # Merge with defaults for any missing networks + for network, url in _DEFAULT_LAB_URLS.items(): + if network not in _NETWORKS: + _NETWORKS[network] = url + + return _NETWORKS + + +def _get_url(network: str) -> str: + """Get Lab base URL for a network.""" + networks = _load_networks() + if network not in networks: + # Try to construct from default pattern + if network in _DEFAULT_LAB_URLS: + return _DEFAULT_LAB_URLS[network] + raise ValueError(f"Unknown network '{network}'. Available: {list(networks.keys())}") + return networks[network] + + +def _fetch_routes() -> dict[str, Any]: + """Fetch routes.json from the configured URL.""" + global _ROUTES + if _ROUTES is not None: + return _ROUTES + + routes_url = os.environ.get( + "ETHPANDAOPS_LAB_ROUTES_URL", + "https://raw.githubusercontent.com/ethpandaops/lab/main/routes.json" + ) + + try: + with httpx.Client(timeout=_TIMEOUT) as client: + resp = client.get(routes_url) + resp.raise_for_status() + _ROUTES = resp.json() + except Exception as e: + # Return default routes structure if fetch fails + _ROUTES = _get_default_routes() + + return _ROUTES + + +def _get_default_routes() -> dict[str, Any]: + """Get default routes structure.""" + return { + "ethereum": [ + { + "id": "ethereum/slots", + "path": "/ethereum/slots/{slot}", + "description": "Ethereum slot details", + "parameters": [{"name": "slot", "type": "integer", "required": True}], + }, + { + "id": "ethereum/epochs", + "path": "/ethereum/epochs/{epoch}", + "description": "Ethereum epoch details", + "parameters": [{"name": "epoch", "type": "integer", "required": True}], + }, + { + "id": "ethereum/validators", + "path": "/ethereum/validators/{validator}", + "description": "Validator details by index or pubkey", + "parameters": [{"name": "validator", "type": "string", "required": True}], + }, + { + "id": "ethereum/blocks", + "path": "/ethereum/blocks/{block}", + "description": "Execution layer block details", + "parameters": [{"name": "block", "type": "string", "required": True}], + }, + { + "id": "ethereum/transactions", + "path": "/ethereum/transactions/{tx_hash}", + "description": "Transaction details", + "parameters": [{"name": "tx_hash", "type": "string", "required": True}], + }, + { + "id": "ethereum/addresses", + "path": "/ethereum/addresses/{address}", + "description": "Address details", + "parameters": [{"name": "address", "type": "string", "required": True}], + }, + { + "id": "ethereum/blobs", + "path": "/ethereum/blobs/{blob_id}", + "description": "Blob details", + "parameters": [{"name": "blob_id", "type": "string", "required": True}], + }, + { + "id": "ethereum/forks", + "path": "/ethereum/forks/{fork_name}", + "description": "Fork information", + "parameters": [{"name": "fork_name", "type": "string", "required": True}], + }, + ] + } + + +def _get_route_by_id(route_id: str) -> dict[str, Any] | None: + """Get route metadata by ID.""" + routes = _fetch_routes() + for category, route_list in routes.items(): + for route in route_list: + if route.get("id") == route_id: + return route + return None + + +def _build_path(path_template: str, params: dict[str, Any]) -> str: + """Build a path from a template and parameters.""" + result = path_template + for key, value in params.items(): + result = result.replace(f"{{{key}}}", str(value)) + return result + + +# Network discovery + + +def list_networks() -> list[dict[str, str]]: + """List networks with Lab explorers.""" + return [{"name": n, "lab_url": u} for n, u in sorted(_load_networks().items())] + + +def get_base_url(network: str) -> str: + """Get Lab base URL for a network.""" + return _get_url(network) + + +# Routes API + + +def get_routes() -> dict[str, Any]: + """Get all Lab routes from routes.json.""" + return _fetch_routes() + + +def get_routes_by_category(category: str) -> list[dict[str, Any]]: + """Get routes filtered by category.""" + routes = _fetch_routes() + return routes.get(category, []) + + +def get_route(route_id: str) -> dict[str, Any] | None: + """Get route metadata by ID.""" + return _get_route_by_id(route_id) + + +# Deep links + + +def link_slot(network: str, slot: int) -> str: + """Generate link to slot page.""" + base = _get_url(network) + route = _get_route_by_id("ethereum/slots") + if route: + path = _build_path(route.get("path", "/ethereum/slots/{slot}"), {"slot": slot}) + return f"{base}{path}" + return f"{base}/ethereum/slots/{slot}" + + +def link_epoch(network: str, epoch: int) -> str: + """Generate link to epoch page.""" + base = _get_url(network) + route = _get_route_by_id("ethereum/epochs") + if route: + path = _build_path(route.get("path", "/ethereum/epochs/{epoch}"), {"epoch": epoch}) + return f"{base}{path}" + return f"{base}/ethereum/epochs/{epoch}" + + +def link_validator(network: str, index_or_pubkey: str) -> str: + """Generate link to validator page.""" + base = _get_url(network) + route = _get_route_by_id("ethereum/validators") + if route: + path = _build_path(route.get("path", "/ethereum/validators/{validator}"), {"validator": index_or_pubkey}) + return f"{base}{path}" + return f"{base}/ethereum/validators/{index_or_pubkey}" + + +def link_block(network: str, number_or_hash: str) -> str: + """Generate link to execution block page.""" + base = _get_url(network) + route = _get_route_by_id("ethereum/blocks") + if route: + path = _build_path(route.get("path", "/ethereum/blocks/{block}"), {"block": number_or_hash}) + return f"{base}{path}" + return f"{base}/ethereum/blocks/{number_or_hash}" + + +def link_address(network: str, address: str) -> str: + """Generate link to address page.""" + base = _get_url(network) + route = _get_route_by_id("ethereum/addresses") + if route: + path = _build_path(route.get("path", "/ethereum/addresses/{address}"), {"address": address}) + return f"{base}{path}" + return f"{base}/ethereum/addresses/{address}" + + +def link_transaction(network: str, tx_hash: str) -> str: + """Generate link to transaction page.""" + base = _get_url(network) + route = _get_route_by_id("ethereum/transactions") + if route: + path = _build_path(route.get("path", "/ethereum/transactions/{tx_hash}"), {"tx_hash": tx_hash}) + return f"{base}{path}" + return f"{base}/ethereum/transactions/{tx_hash}" + + +def link_blob(network: str, blob_id: str) -> str: + """Generate link to blob page.""" + base = _get_url(network) + route = _get_route_by_id("ethereum/blobs") + if route: + path = _build_path(route.get("path", "/ethereum/blobs/{blob_id}"), {"blob_id": blob_id}) + return f"{base}{path}" + return f"{base}/ethereum/blobs/{blob_id}" + + +def link_fork(network: str, fork_name: str) -> str: + """Generate link to fork information page.""" + base = _get_url(network) + route = _get_route_by_id("ethereum/forks") + if route: + path = _build_path(route.get("path", "/ethereum/forks/{fork_name}"), {"fork_name": fork_name}) + return f"{base}{path}" + return f"{base}/ethereum/forks/{fork_name}" + + +def build_url(network: str, route_id: str, params: dict[str, Any]) -> str: + """Build URL for a specific route with parameters. + + Args: + network: Network name (e.g., "mainnet", "sepolia") + route_id: Route identifier (e.g., "ethereum/slots") + params: Dictionary of parameters to substitute in the route path + + Returns: + Complete URL for the route + + Raises: + ValueError: If network or route is not found + """ + base = _get_url(network) + route = _get_route_by_id(route_id) + + if not route: + raise ValueError(f"Unknown route '{route_id}'") + + path_template = route.get("path", "") + if not path_template: + raise ValueError(f"Route '{route_id}' has no path defined") + + path = _build_path(path_template, params) + return f"{base}{path}"