46 lines
1.9 KiB
Python
46 lines
1.9 KiB
Python
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
from typing import Any
|
|
|
|
from kab_ingestion.github import GitHubConnector
|
|
from kab_ingestion.models import IngestionRequest, IngestionResult
|
|
from kab_ingestion.sharepoint import SharePointConnector
|
|
from .card import AGENT_CARD, AgentCard
|
|
from .publication import publish_document
|
|
from .settings import Settings, get_settings
|
|
from .validation import validate_document
|
|
|
|
|
|
@dataclass
|
|
class IngestionService:
|
|
settings: Settings
|
|
scm: Any
|
|
cms: Any = None
|
|
|
|
def ingest(self, request: IngestionRequest) -> list[IngestionResult]:
|
|
from .workflows import create_ingestion_workflow
|
|
from .states.ingestion_state import IngestionState
|
|
|
|
workflow = create_ingestion_workflow(
|
|
scm=self.scm,
|
|
cms=self.cms,
|
|
allowed_sources=self.settings.allowed_sources,
|
|
publish_enabled=self.settings.publish_enabled,
|
|
require_governance_approval=self.settings.require_governance_approval,
|
|
)
|
|
initial_state: IngestionState = {
|
|
"request": request,
|
|
"tenant_id": request.metadata.get("tenant_id", "default_tenant"),
|
|
}
|
|
final_state = workflow.invoke(initial_state)
|
|
return final_state.get("published_results", [])
|
|
|
|
|
|
def create_agent_app(settings: Settings | None = None) -> dict[str, Any]:
|
|
config = settings or get_settings()
|
|
scm = GitHubConnector(config.github_token, config.github_api_base)
|
|
cms = SharePointConnector(config.sharepoint_site_url, config.sharepoint_access_token) if config.sharepoint_site_url and config.sharepoint_access_token else None
|
|
service = IngestionService(config, scm, cms)
|
|
return {"card": AGENT_CARD, "service": service, "routes": {"/health": "health", "/card": "card", "/ingest": "ingest"}, "triggers": {"github_push": "github_trigger", "manual_ingest": "manual_trigger"}}
|