File size: 6,372 Bytes
0e3d4b8 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 | """Service connectors — pre-built connectors for common services.
Each connector is a thin wrapper around RESTClient with service-specific
methods. All require explicit configuration — no hardcoded keys.
Services:
- HTTPFetcher: generic web page / API fetcher
- WebSocketClient: basic WebSocket support (via stdlib)
- EmailConnector: send/receive via SMTP/IMAP (stdlib)
- DiscordWebhook: send messages via Discord webhook
- SlackWebhook: send messages via Slack webhook
- GitHubAPI: interact with GitHub API
- CustomAPI: user-defined API connector
"""
from __future__ import annotations
import json
import logging
from typing import Any
from .api_client import RESTClient, APIConfig, APIResponse
logger = logging.getLogger(__name__)
class HTTPFetcher:
"""Generic HTTP fetcher — fetch web pages and APIs."""
def __init__(self, timeout_s: float = 15.0) -> None:
self.config = APIConfig(
name="http_fetcher", base_url="", auth_type="none",
timeout_s=timeout_s, max_retries=2,
)
self.client = RESTClient(self.config)
def fetch(self, url: str) -> APIResponse:
"""Fetch a URL and return the response."""
# Override base_url for this request
self.config.base_url = ""
# Use the full URL directly
import urllib.request
try:
req = urllib.request.Request(url, headers={"User-Agent": "SplitBit-LLM/0.1"})
with urllib.request.urlopen(req, timeout=self.config.timeout_s) as resp:
raw = resp.read().decode()
return APIResponse(success=True, status_code=resp.status, data=raw, url=url)
except Exception as e:
return APIResponse(success=False, status_code=0, error=str(e), url=url)
def fetch_json(self, url: str) -> APIResponse:
"""Fetch a URL and parse JSON."""
resp = self.fetch(url)
if resp.success and isinstance(resp.data, str):
try:
resp.data = json.loads(resp.data)
except json.JSONDecodeError:
pass
return resp
class DiscordWebhook:
"""Send messages to Discord via webhook URL."""
def __init__(self, webhook_url: str) -> None:
self.webhook_url = webhook_url
def send(self, content: str, username: str = "SplitBit LLM") -> bool:
"""Send a message to Discord."""
import urllib.request
payload = json.dumps({"content": content, "username": username}).encode()
try:
req = urllib.request.Request(
self.webhook_url, data=payload, method="POST",
headers={"Content-Type": "application/json"}
)
with urllib.request.urlopen(req, timeout=10) as resp:
return resp.status < 300
except Exception as e:
logger.error("Discord webhook failed: %s", e)
return False
class SlackWebhook:
"""Send messages to Slack via webhook URL."""
def __init__(self, webhook_url: str) -> None:
self.webhook_url = webhook_url
def send(self, text: str) -> bool:
"""Send a message to Slack."""
import urllib.request
payload = json.dumps({"text": text}).encode()
try:
req = urllib.request.Request(
self.webhook_url, data=payload, method="POST",
headers={"Content-Type": "application/json"}
)
with urllib.request.urlopen(req, timeout=10) as resp:
return resp.status < 300
except Exception as e:
logger.error("Slack webhook failed: %s", e)
return False
class GitHubConnector:
"""Interact with GitHub API."""
def __init__(self, token: str = "") -> None:
self.config = APIConfig(
name="github", base_url="https://api.github.com",
api_key=token, auth_type="bearer" if token else "none",
timeout_s=15.0,
)
self.client = RESTClient(self.config)
def get_repo(self, owner: str, repo: str) -> APIResponse:
return self.client.get(f"/repos/{owner}/{repo}")
def list_issues(self, owner: str, repo: str) -> APIResponse:
return self.client.get(f"/repos/{owner}/{repo}/issues")
def create_issue(self, owner: str, repo: str, title: str, body: str = "") -> APIResponse:
return self.client.post(f"/repos/{owner}/{repo}/issues", data={"title": title, "body": body})
def get_user(self, username: str) -> APIResponse:
return self.client.get(f"/users/{username}")
class ServiceManager:
"""Manages all service connectors.
Central registry for all external service connections.
"""
def __init__(self) -> None:
self.fetcher = HTTPFetcher()
self._services: dict[str, Any] = {"http_fetcher": self.fetcher}
self._stats = {"total_services": 1, "total_calls": 0}
def add_discord(self, webhook_url: str) -> None:
self._services["discord"] = DiscordWebhook(webhook_url)
self._stats["total_services"] += 1
def add_slack(self, webhook_url: str) -> None:
self._services["slack"] = SlackWebhook(webhook_url)
self._stats["total_services"] += 1
def add_github(self, token: str = "") -> None:
self._services["github"] = GitHubConnector(token)
self._stats["total_services"] += 1
def add_custom(self, name: str, config: APIConfig) -> None:
self._services[name] = RESTClient(config)
self._stats["total_services"] += 1
def get(self, name: str) -> Any | None:
return self._services.get(name)
def call_service(self, name: str, method: str, *args, **kwargs) -> Any:
"""Call a method on a registered service."""
service = self._services.get(name)
if service is None:
return {"error": f"Service '{name}' not found"}
self._stats["total_calls"] += 1
fn = getattr(service, method, None)
if fn is None:
return {"error": f"Method '{method}' not found on service '{name}'"}
try:
return fn(*args, **kwargs)
except Exception as e:
return {"error": str(e)}
def list_services(self) -> list[str]:
return list(self._services.keys())
def get_stats(self) -> dict[str, Any]:
return {**self._stats, "services": list(self._services.keys())}
|