Skip to content

API Reference: Federation

This document covers the FCC federation subsystem, which provides cross-project entity resolution, namespace management, change tracking, and a top-level registry for coordinating federated projects across the 11-project ecosystem.

flowchart LR
    subgraph Project A
        EA[Entity: RC]
    end
    subgraph Project B
        EB[Entity: researcher]
    end
    EA -->|VocabularyMapping| ER[EntityResolver]
    ER --> NR[NamespaceRegistry]
    NR -->|resolve URI| FE[FederatedEntity]
    EB -->|VocabularyMapping| ER
    ER --> CT[ChangeTracker]
    CT -->|audit| CS[ChangeSet]
    FE --> FR[FederationRegistry]

EntityResolver

fcc.federation.entities.EntityResolver resolves local entity identifiers across namespaces using vocabulary mappings.

Setup

from fcc.federation.entities import EntityResolver, FederatedEntity
from fcc.objectmodel.mapping import VocabularyMapping

resolver = EntityResolver()

# Register vocabulary mappings
resolver.add_mapping(VocabularyMapping(
    source_id="RC",
    source_name="Research Coordinator",
    source_vocabulary="fcc",
    target_id="researcher",
    target_name="Researcher",
    target_vocabulary="cto",
    similarity_score=0.85,
))

Resolution

# Resolve a single entity
result = resolver.resolve("RC", "fcc", "cto")
if result is not None:
    print(f"Resolved: {result.canonical_id}")  # "cto:researcher"
    print(f"Confidence: {result.confidence}")   # 0.85

# Batch resolution
results = resolver.resolve_batch(
    ids=[("RC", "fcc"), ("SA", "fcc"), ("WA", "fcc")],
    target_namespace="cto",
)
for entity in results:
    print(f"{entity.local_id}: resolved={entity.resolved}")

print(resolver.mapping_count())  # Number of registered mappings

FederatedEntity

Frozen dataclass representing a resolved (or pending) cross-namespace entity:

Field Type Description
local_id str Identifier within the source namespace
namespace str Namespace the entity belongs to
canonical_id str Fully qualified namespace:local_id identifier
resolved bool Whether resolution has been completed
confidence float Resolution confidence score (0.0--1.0)
source_vocabulary str Vocabulary the entity was originally expressed in
properties dict Arbitrary metadata

NamespaceRegistry

fcc.federation.namespaces.NamespaceRegistry manages namespace configurations for federated projects with O(1) lookup and URI resolution.

Setup

from fcc.federation.namespaces import NamespaceRegistry, NamespaceConfig

# Load default registry (11 ecosystem namespaces)
registry = NamespaceRegistry.default_registry()

# Or build manually
registry = NamespaceRegistry()
registry.register(NamespaceConfig(
    namespace="fcc",
    prefix="fcc",
    base_uri="https://fcc.example.org/ontology/",
    version="0.8.0",
    description="FCC Agent Team Framework",
))

Operations

# Retrieve a namespace
config = registry.get("fcc")
print(config.base_uri)  # "https://fcc.example.org/ontology/"

# Resolve a URI
uri = registry.resolve_uri("fcc", "RC")
# "https://fcc.example.org/ontology/RC"

# Validate namespace existence
assert registry.validate("fcc")

# List all
all_configs = registry.all_namespaces()
print(registry.count())  # 11

Loading from YAML

from pathlib import Path

registry = NamespaceRegistry.from_yaml(Path("namespace_registry.yaml"))

The YAML file must have a top-level namespaces key with a list of namespace configurations.

ChangeTracker

fcc.federation.change_tracking.ChangeTracker tracks model changes and produces change sets for federation auditing.

Tracking Changes

from fcc.federation.change_tracking import ChangeTracker, ModelChange

tracker = ChangeTracker()

# Record individual changes
tracker.track_change(ModelChange(
    entity_id="persona_RC",
    change_type="modified",
    namespace="fcc",
    old_value={"role": "Old role"},
    new_value={"role": "Updated role"},
))

tracker.track_change(ModelChange(
    entity_id="persona_NEW",
    change_type="added",
    namespace="fcc",
    new_value={"name": "New Persona"},
))

Creating Change Sets

# Bundle pending changes into a ChangeSet
changeset = tracker.create_changeset(
    description="Phase 13 persona updates",
    source_namespace="fcc",
)

print(changeset.added_count)     # 1
print(changeset.modified_count)  # 1
print(changeset.removed_count)   # 0

# Pending queue is cleared after creating a changeset

Diffing Facades

# Compare two ModelFacade stats to generate a ChangeSet
changeset = tracker.diff_facades(old_facade, new_facade)
for change in changeset.changes:
    print(f"{change.entity_id}: {change.change_type}")

History

# View all recorded change sets
for entry in tracker.history():
    cs = entry.change_set
    print(f"{cs.description}: {len(cs.changes)} changes")
    print(f"  Applied: {entry.applied}")

# Reset
tracker.clear()

ModelChange

Frozen dataclass with entity_id, change_type (one of "added", "modified", "removed"), namespace, old_value, new_value, and timestamp.

ChangeSet

Frozen dataclass bundling multiple ModelChange instances with source_namespace, description, and timestamp. Properties: added_count, modified_count, removed_count.

FederationRegistry

fcc.federation.registry.FederationRegistry is the central coordination point aggregating NamespaceRegistry, EntityResolver, and ChangeTracker.

Setup

from fcc.federation.registry import FederationRegistry
from fcc.federation.namespaces import NamespaceConfig

federation = FederationRegistry()

# Register projects
federation.add_project(
    "fcc",
    facade=fcc_facade,
    namespace_config=NamespaceConfig(
        namespace="fcc",
        prefix="fcc",
        base_uri="https://fcc.example.org/ontology/",
    ),
)
federation.add_project("cto", facade=cto_facade)
federation.add_project("aome", facade=aome_facade)

Cross-Project Resolution

# Resolve an entity across all registered projects
results = federation.resolve_across_projects("RC", "fcc")
for entity in results:
    print(f"  -> {entity.namespace}:{entity.local_id} (conf={entity.confidence})")

Health Check

health = federation.health_check()
print(f"Projects: {health['project_count']}")
print(f"Namespaces: {health['namespace_count']}")
print(f"Mappings: {health['mapping_count']}")

for project in health["projects"]:
    print(f"  {project['namespace']}: facade={project['has_facade']}")

Accessing Sub-Components

ns_registry = federation.namespace_registry
entity_resolver = federation.entity_resolver
change_tracker = federation.change_tracker

# Add mappings through the resolver
entity_resolver.add_mapping(mapping)

# Track changes
change_tracker.track_change(change)

# View change log
for entry in federation.change_log():
    print(entry.change_set.description)

# List projects
print(federation.all_projects())     # ["aome", "cto", "fcc"]
print(federation.project_count())    # 3

CrossProjectAssessment

fcc.objectmodel.federation.assess_cross_project() provides a high-level assessment of cross-project alignment.

from fcc.objectmodel.federation import assess_cross_project

assessment = assess_cross_project(fcc_facade, cto_facade)

Import Paths Summary

Class Import Path
EntityResolver fcc.federation.entities
FederatedEntity fcc.federation.entities
NamespaceRegistry fcc.federation.namespaces
NamespaceConfig fcc.federation.namespaces
ChangeTracker fcc.federation.change_tracking
ModelChange fcc.federation.change_tracking
ChangeSet fcc.federation.change_tracking
FederationRegistry fcc.federation.registry
assess_cross_project fcc.objectmodel.federation

See Also