0987704264
GCAP-SPEC-CONNECTOR-DESCRIPTOR-0001 implementation.
ConnectorPlugin — abstract base for governed connectors.
ConnectorRegistry — register/discover connectors.
ConnectorRuntime — wraps invoke with Chronicle lineage.
EchoConnector — dev/test example.
API endpoints:
GET /connectors/ — browse catalogue
GET /connectors/{id}/ — connector descriptor
POST /connectors/{id}/invoke/ — governed invocation
GET /connectors/{id}/health/ — health check
5 tests passing.
Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
65 lines
1.9 KiB
Python
65 lines
1.9 KiB
Python
"""ConnectorPlugin ABC per GCAP-SPEC-CONNECTOR-DESCRIPTOR-0001 section 4."""
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
from abc import ABC, abstractmethod
|
|
from dataclasses import dataclass, field
|
|
from typing import Any
|
|
|
|
|
|
@dataclass
|
|
class ConnectorContext:
|
|
chronicle_session_id: str = ""
|
|
gsap_context_id: str = ""
|
|
credentials: dict[str, Any] = field(default_factory=dict)
|
|
pipeline_run_id: str = ""
|
|
dag_id: str = ""
|
|
|
|
|
|
@dataclass
|
|
class ConnectorResult:
|
|
success: bool = False
|
|
data: Any = None
|
|
error: str | None = None
|
|
lineage_cid: str = ""
|
|
metadata: dict[str, Any] = field(default_factory=dict)
|
|
|
|
|
|
class ConnectorPlugin(ABC):
|
|
"""Abstract base for governed connectors."""
|
|
|
|
connector_id: str = ""
|
|
corpus_entry_cid: str = ""
|
|
capability_mask: int = 0
|
|
declared_endpoints: list[str] = []
|
|
accord_template: str = ""
|
|
gsap_required: bool = True
|
|
chronicle_enabled: bool = True
|
|
|
|
@abstractmethod
|
|
async def invoke(
|
|
self, operation: str, parameters: dict[str, Any], context: ConnectorContext
|
|
) -> ConnectorResult:
|
|
...
|
|
|
|
@abstractmethod
|
|
def health_check(self) -> bool:
|
|
...
|
|
|
|
def compute_parameters_cid(self, parameters: dict[str, Any]) -> str:
|
|
canonical = json.dumps(parameters, sort_keys=True, separators=(",", ":"))
|
|
return "sha256:" + hashlib.sha256(canonical.encode()).hexdigest()
|
|
|
|
def descriptor(self) -> dict[str, Any]:
|
|
return {
|
|
"@context": "https://schema.gsap.dev/connector/v1",
|
|
"@type": "ConnectorDescriptor",
|
|
"connector_id": self.connector_id,
|
|
"corpus_entry_cid": self.corpus_entry_cid,
|
|
"capability_mask": self.capability_mask,
|
|
"declared_endpoints": self.declared_endpoints,
|
|
"accord_template": self.accord_template,
|
|
"gsap_required": self.gsap_required,
|
|
"chronicle_enabled": self.chronicle_enabled,
|
|
}
|