Ecosystem Integration Guide¶
This guide covers how the FCC Agent Team Framework integrates with the broader ecosystem of projects, how external projects consume FCC primitives, and how to build new integrations using the clean-room adapter pattern established by the CTO bridge.
1. Ecosystem Overview¶
The FCC framework sits at the center of an 11-project ecosystem. Each project occupies one of four adoption tiers relative to FCC:
| Tier | Meaning |
|---|---|
| Authority | Owns and defines the FCC/R.I.S.C.E.A.R. specification |
| Producer | Actively produces FCC artifacts and consumes the framework |
| Consumer | Uses FCC framework but does not produce extensive artifacts |
| Observer | Referenced in ecosystem but minimal FCC integration |
Project Registry¶
| Project | Role | Version | FCC Tier | A2A Status | MCP Status | Tests |
|---|---|---|---|---|---|---|
| FCC Agent Team Ext | Canonical authority | v0.6.0 | Authority | -- | -- | 2,932 |
| PAOM | Primary consumer, 13 plugin personas, SENTINEL routing | v3.0.0 | Producer | Agent Card | 25 MCP tools, 7 output formats | 3,119 |
| AOME | Vocabulary taxonomy, 5 plugin personas, privacy ontology (1,026 nodes) | v0.1.0 | Producer | Agent Card builder | 9 MCP tools + 5 resources | 2,455 |
| CONSTEL | TMF knowledge graph, 5 plugin personas, UCM metadata | v0.2.0 | Producer | Agent Card builder with routes | MCP via FastAPI | 1,277 |
| CTO | Object model foundation, telecom ontology | v6.1.0 | Consumer | Full A2A (CapabilityCard) | Integrated MCP server | 710 |
| Research Center | Statistical foundation, experiment harness | v0.1.0 | Consumer | Full A2A (8 skills) | 4 MCP tools | 848 |
| Sky-Parlour | D3.js visualization, 218 workflows, React 18 | PI 1 | Producer | A2A handlers | Extended MCP client | 5,857 |
| AI COE Docs | FCC artifact producer, patent evaluation, Ralph Loop | -- | Producer | A2A server | MCP server | 1,582 |
| Distiller (Fornax) | Vocabulary distillation, Horologium naming | -- | Producer | -- | -- | 4,960 |
| P1/FlowX | Event capture, blueprint-first governance | -- | Observer | -- | OTel integration | 1,077 |
| L2 Distro Hub | Release asset management and distribution | -- | Observer | -- | -- | -- |
Port Allocations¶
Ecosystem projects coordinate port allocations to avoid conflicts:
PAOM :9001 (API), :9010 (MCP)
AOME :8001 (API), :8100 (MCP), :8501 (Streamlit)
CONSTEL :8100 (FastAPI)
Sky-Parlour :5173 (frontend), :5174 (prototypes)
AI COE Docs :8200 (A2A), :8201 (docs), :8270 (dashboards), :8271 (Streamlit)
Distiller :5173 (frontend), :8002 (backend)
P1 :3001 (event-collector), :5174-5177 (UIs)
Research Ctr :3300 (MCP)
The canonical registry lives at src/fcc/data/ecosystem/project_registry.yaml.
All port conflict resolutions are tracked in that file under
port_conflicts_resolved.
2. Clean-Room Integration Philosophy¶
FCC never imports external project code directly. Instead, every integration follows a four-step Protocol -- Dataclass -- Adapter -- Sentinel pattern that keeps the framework free of hard dependencies while still enabling rich interoperability.
Step 1: Define typing.Protocol interfaces¶
Protocols establish the structural contract that both FCC's internal implementations and external adapters must satisfy. Because Python protocols use structural subtyping, the external project does not need to inherit from any FCC base class.
# src/fcc/objectmodel/base.py
from typing import Protocol, TypeVar, runtime_checkable
from collections.abc import Iterator
T_co = TypeVar("T_co", covariant=True)
@runtime_checkable
class DomainEntity(Protocol):
"""Minimal contract for a domain entity."""
@property
def id(self) -> int | str: ...
def to_dict(self) -> dict[str, Any]: ...
@runtime_checkable
class RepositoryProtocol(Protocol[T_co]):
"""Generic repository with O(1) lookup semantics."""
def add(self, entity: Any) -> None: ...
def get_by_id(self, id: int | str) -> T_co | None: ...
def get_all(self) -> list[Any]: ...
def count(self) -> int: ...
def __iter__(self) -> Iterator[Any]: ...
def __len__(self) -> int: ...
def __contains__(self, id: int | str) -> bool: ...
Why protocols? Both FCC's frozen=True dataclasses and CTO's slots=True
dataclasses satisfy these contracts without sharing an inheritance hierarchy.
Any external model that has .id and .to_dict() works automatically.
Step 2: Use frozen dataclasses for data transfer¶
Data transfer objects are @dataclass(frozen=True) instances that cross
module boundaries. Immutability prevents downstream code from silently
mutating shared state.
# src/fcc/objectmodel/mapping.py
from dataclasses import dataclass, field
from datetime import datetime, timezone
@dataclass(frozen=True)
class VocabularyMapping:
"""A semantic mapping between entities in two vocabularies."""
source_id: str
source_name: str
source_vocabulary: str
target_id: str
target_name: str
target_vocabulary: str
similarity_score: float
mapping_rank: int = 0
requires_review: bool = field(init=False)
created_at: datetime = field(
default_factory=lambda: datetime.now(timezone.utc)
)
def __post_init__(self) -> None:
if not (0.0 <= self.similarity_score <= 1.0):
raise ValueError(f"similarity_score must be 0.0-1.0")
object.__setattr__(
self, "requires_review", self.similarity_score < 0.75
)
Key design choices:
frozen=Trueensures immutability.- Computed fields use
object.__setattr__in__post_init__because the dataclass is frozen. to_dict()andfrom_dict()classmethods provide serialization without requiring any serialization library.
Step 3: Build adapter classes bridging protocols to implementations¶
Adapters translate external classes into FCC protocol-compatible objects. They delegate every call to the underlying implementation without copying data, preserving original performance characteristics.
# src/fcc/objectmodel/cto_bridge.py
class CTORepositoryAdapter:
"""Wraps a CTO Repository[T] to satisfy RepositoryProtocol."""
def __init__(self, cto_repo: Any) -> None:
if not cto_available():
raise ImportError("cto package is not installed")
self._repo = cto_repo
def add(self, entity: Any) -> None:
self._repo.add(entity)
def get_by_id(self, id: int | str) -> Any | None:
return self._repo.get_by_id(id)
def get_all(self) -> list[Any]:
return self._repo.get_all()
def count(self) -> int:
return self._repo.count()
# ... __iter__, __len__, __contains__ delegate likewise
The adapter is verified at class definition time:
Step 4: Use availability sentinel functions for optional dependencies¶
Sentinel functions cache the result of an import attempt so that calling
code can branch cheaply without repeated try/except blocks.
_CTO_AVAILABLE: bool | None = None
def cto_available() -> bool:
"""Return True if the cto package can be imported."""
global _CTO_AVAILABLE
if _CTO_AVAILABLE is None:
try:
import cto # noqa: F401
_CTO_AVAILABLE = True
except ImportError:
_CTO_AVAILABLE = False
return _CTO_AVAILABLE
Usage at call sites:
from fcc.objectmodel.cto_bridge import cto_available, CTOFacadeAdapter
if cto_available():
from cto.core.model import CTOModel
facade = CTOFacadeAdapter(CTOModel(...))
else:
# fall back to FCC-native ExampleFacade or skip CTO features
from fcc.objectmodel.examples import create_sample_model
facade = create_sample_model()
Summary of the Pattern¶
Step 1 Step 2 Step 3 Step 4
-------- ---------------- ----------------- ---------------
Protocol Frozen Dataclass Adapter Class Sentinel fn()
-------- ---------------- ----------------- ---------------
| | | |
| structural | data transfer | delegation to | guards
| contract | objects | external impl | against
| (no import) | (immutable) | (wraps 3rd-party) | ImportError
v v v v
3. CTO Bridge as Canonical Pattern¶
The CTO bridge (src/fcc/objectmodel/cto_bridge.py) is the reference
implementation for integrating any external object model. It demonstrates all
four steps of the clean-room pattern in a single, self-contained module.
Module Structure¶
src/fcc/objectmodel/
base.py # Step 1: DomainEntity, RepositoryProtocol
facade.py # Step 1: ModelFacade ABC (stats, search, get_full)
mapping.py # Step 2: VocabularyMapping, MappingStore protocol
evolution.py # Step 2: EvolutionStage, ObjectModelAssessment
cto_bridge.py # Steps 3+4: CTOFacadeAdapter, CTORepositoryAdapter, cto_available()
examples.py # Reference: ExampleRepository, ExampleFacade
cto_available() Sentinel¶
from fcc.objectmodel.cto_bridge import cto_available
# Call this BEFORE attempting any CTO import
if cto_available():
# CTO is installed -- use the real adapter
...
else:
# CTO is not installed -- degrade gracefully
...
The sentinel result is cached globally after the first call. For tests, use
_reset_cto_sentinel() to clear the cache.
CTOFacadeAdapter¶
Bridges CTO's CTOModel (1,520 concepts, 6,697 elements, 2,463
associations) behind FCC's ModelFacade ABC:
from fcc.objectmodel.cto_bridge import cto_available, CTOFacadeAdapter
if cto_available():
from cto.core.model import CTOModel
cto_model = CTOModel(version="6.1.0")
facade = CTOFacadeAdapter(cto_model)
# ModelFacade interface
print(facade.name) # "CTO"
print(facade.version) # "6.1.0"
print(facade.stats()) # {"concepts": 1520, ...}
results = facade.search("Bearer") # list of CTO concepts
full = facade.get_full("concept-42") # dict with related objects
# CTO-specific: access wrapped repositories
concepts_repo = facade.get_repository("concepts")
if concepts_repo is not None:
print(concepts_repo.count()) # 1520
CTORepositoryAdapter¶
Wraps any CTO Repository[T] to satisfy RepositoryProtocol[T]:
from fcc.objectmodel.cto_bridge import CTORepositoryAdapter
adapter = CTORepositoryAdapter(cto_model.concepts)
# All RepositoryProtocol methods are available
entity = adapter.get_by_id("concept-42")
all_entities = adapter.get_all()
print(len(adapter)) # 1520
print("concept-42" in adapter) # True
Writing Your Own Bridge¶
To integrate a new external project (e.g., AOME's privacy taxonomy), follow the same pattern:
# src/fcc/objectmodel/aome_bridge.py (hypothetical)
_AOME_AVAILABLE: bool | None = None
def aome_available() -> bool:
global _AOME_AVAILABLE
if _AOME_AVAILABLE is None:
try:
import aome # noqa: F401
_AOME_AVAILABLE = True
except ImportError:
_AOME_AVAILABLE = False
return _AOME_AVAILABLE
class AOMEFacadeAdapter(ModelFacade):
"""Bridges AOME's TaxonomyModel behind FCC's ModelFacade."""
def __init__(self, aome_model: Any) -> None:
if not aome_available():
raise ImportError("aome package is not installed")
self._model = aome_model
super().__init__(
name=getattr(aome_model, "name", "AOME"),
version=getattr(aome_model, "version", "unknown"),
)
def stats(self) -> dict[str, int]:
return self._model.stats()
def search(self, query: str, *, limit: int = 50) -> list[Any]:
return self._model.search_nodes(query, limit=limit)
def get_full(self, entity_id: int | str) -> dict[str, Any] | None:
return self._model.get_node_full(entity_id)
4. A2A Skill Mapping¶
The FCC R.I.S.C.E.A.R. specification maps directly to Google's Agent-to-Agent (A2A) protocol's Agent Card structure. Each FCC persona can be exposed as an A2A agent.
R.I.S.C.E.A.R. to Agent Card Mapping¶
| R.I.S.C.E.A.R. Component | A2A Agent Card Field | Example |
|---|---|---|
| Role | name, description |
"Research Curator", "Finds and validates research sources" |
| Role Skills | skills[] |
[{"id": "source-validation", "name": "Source Validation", ...}] |
| Style | defaultInputModes, defaultOutputModes |
["text/plain"], ["application/json"] |
| Constraints | skills[].tags[] |
["max-sources-10", "peer-reviewed-only"] |
| Expected Output | skills[].outputModes |
["application/json", "text/markdown"] |
| Archetype | skills[].description |
"The Librarian — systematic, thorough, citation-focused" |
| Responsibilities | skills[].description (expanded) |
Mapped to skill descriptions |
| Role Collaborators | Agent Card url cross-references |
Links to collaborating agent endpoints |
| Cross-Reference Matrix | A2A task routing patterns | Upstream/downstream maps to task delegation |
Generating an Agent Card from a Persona¶
from fcc.personas.registry import PersonaRegistry
from fcc._resources import get_data_dir
registry = PersonaRegistry.from_yaml_directory(get_data_dir() / "personas")
persona = registry.get("RC") # Research Curator
agent_card = {
"name": persona.name,
"description": persona.riscear.role,
"url": f"https://api.example.com/agents/{persona.id.lower()}",
"version": "1.0.0",
"capabilities": {
"streaming": False,
"pushNotifications": False,
},
"skills": [
{
"id": f"{persona.id.lower()}-{skill.lower().replace(' ', '-')}",
"name": skill,
"description": f"{persona.riscear.archetype} — {skill}",
"tags": [persona.category, persona.fcc_phase.lower()],
"inputModes": ["text/plain", "application/json"],
"outputModes": ["application/json"],
}
for skill in persona.riscear.role_skills
],
"defaultInputModes": ["text/plain"],
"defaultOutputModes": ["application/json"],
}
Cross-Reference Matrix as Collaboration Patterns¶
The CrossReferenceMatrix defines how personas interact. In an A2A
deployment, these translate to task routing rules:
from fcc.personas.cross_reference import CrossReferenceMatrix
matrix = CrossReferenceMatrix.from_yaml(
get_data_dir() / "personas" / "cross_reference.yaml"
)
# Downstream targets = agents this persona delegates tasks to
downstream = matrix.downstream("RC")
for entry in downstream:
print(f" RC -> {entry.target_id}: {entry.relationship_type}")
# e.g., RC -> BC: handoff (Research Curator hands off to Build Coordinator)
# Upstream sources = agents that delegate tasks to this persona
upstream = matrix.upstream("BC")
for entry in upstream:
print(f" {entry.source_id} -> BC: {entry.relationship_type}")
Ecosystem A2A Deployment Map¶
A2A Agent Cards
===============
PAOM Agent Card CTO CapabilityCard
+-----------------+ +------------------+
| SENTINEL Router | | 1,520 concepts |
| 13 personas | | Full A2A skills |
| skills: [...] | | skills: [...] |
+-----------------+ +------------------+
| |
v v
+-------------------------------+
| FCC Cross-Reference Matrix |
| (persona collaboration map) |
+-------------------------------+
| |
v v
AOME Agent Card Research Center Card
+-----------------+ +------------------+
| Agent Card bldr | | 8 A2A skills |
| 5 personas | | Statistical ops |
| 9 MCP tools | | 4 MCP tools |
+-----------------+ +------------------+
5. MCP Tool Mapping¶
FCC CLI commands map to Model Context Protocol (MCP) tool definitions. Consumer projects (PAOM, AOME, Research Center) expose these as MCP tools for LLM-driven workflows.
CLI-to-MCP Mapping Table¶
| FCC CLI Command | MCP Tool Name | Description | Input Schema |
|---|---|---|---|
fcc validate |
validate_project |
Validate FCC project structure | {"directory": "string"} |
fcc generate-docs |
generate_docs |
Generate docs-as-code from persona specs | {"directory": "string", "personas": "string"} |
fcc simulate |
run_simulation |
Run an FCC workflow simulation | {"scenario": "string", "mock": "boolean"} |
fcc action run |
execute_action |
Run a persona workflow action | {"persona": "string", "action_type": "string"} |
fcc plugins list |
list_plugins |
List all discovered FCC plugins | {} |
fcc plugins validate |
validate_plugins |
Validate all discovered plugins | {} |
fcc collab start |
start_collaboration |
Start a collaboration session | {"workflow": "string", "participants": "string[]"} |
fcc dashboard personas |
get_persona_catalog |
Retrieve persona catalog data | {"category": "string?"} |
Example MCP Tool Definition¶
An MCP server exposing FCC capabilities would define tools like this:
{
"name": "run_simulation",
"description": "Run an FCC Find-Create-Critique workflow simulation with the specified scenario. Returns trace data including step count and AI call count.",
"inputSchema": {
"type": "object",
"properties": {
"scenario": {
"type": "string",
"description": "Scenario ID (e.g. GEN-001)",
"default": "GEN-001"
},
"mock": {
"type": "boolean",
"description": "Use mock AI client (true) or real AI (false)",
"default": true
}
},
"required": []
}
}
MCP Tool Implementation Pattern¶
# Hypothetical MCP server wrapping FCC CLI logic
from fcc.personas.registry import PersonaRegistry
from fcc.workflow.action_engine import ActionEngine
from fcc.workflow.actions import WorkflowActionRegistry, WorkflowActionType
async def handle_execute_action(params: dict) -> dict:
"""MCP tool handler for execute_action."""
persona_id = params["persona"].upper()
action_type = WorkflowActionType(params["action_type"])
data_dir = get_data_dir()
persona_reg = PersonaRegistry.from_yaml_directory(data_dir / "personas")
action_reg = WorkflowActionRegistry.from_yaml_directory(
data_dir / "personas" / "actions"
)
engine = ActionEngine(persona_reg, action_reg)
result = engine.run(persona_id, action_type)
return {
"success": result.success,
"content": result.content if result.success else None,
"error": result.error if not result.success else None,
}
Ecosystem MCP Deployment¶
Consumer/LLM
|
| MCP protocol (JSON-RPC over stdio/SSE)
v
+------------------+ +------------------+ +------------------+
| PAOM MCP :9010 | | AOME MCP :8100 | | ResearchCtr :3300|
| 25 tools | | 9 tools | | 4 tools |
| 7 output formats | | 5 resources | | |
+------------------+ +------------------+ +------------------+
| | |
+-------------------------+-------------------------+
|
FCC Framework (pip install fcc)
PersonaRegistry, ActionEngine,
SimulationEngine, EventBus, ...
6. Cross-Project Data Flow¶
Primary Data Flow: FCC to Consumers¶
+--------------------+
| FCC Agent Team Ext |
| (Authority) |
| |
| Personas (84+23) |
| Workflows (6) |
| Schemas (10) |
| Actions (204) |
+--------+-----------+
|
| pip install fcc
| (Python package)
|
+----+----+----+----+----+
| | | |
v v v v
+------+ +------+ +------+ +----------+
| PAOM | | AOME | |CONSTEL| | AI COE |
| | | | | | | Docs |
+--+---+ +--+---+ +--+---+ +----+-----+
| | | |
+---------+---------+------------+
|
v
+---------+---------+
| Plugin Ecosystem |
| (entry points) |
+-------------------+
Vocabulary and Taxonomy Flow¶
+--------+ taxonomy +--------+ metadata +---------+
| AOME | --------------> |CONSTEL | --------------> | PAOM |
| 1,026 | vocabulary | TMF KG | enrichment | SENTINEL|
| privacy| | | | routing |
| nodes | | | | |
+---+----+ +---+----+ +----+----+
| | |
| privacy | knowledge | LLM
| classifications | graph links | routing
| | |
v v v
+--------+ +--------+ +----------+
|Research| | CTO | |Sky-Parlour|
|Center | | 1,520 | | D3.js vis |
|stats | |concepts| | 218 flows |
+--------+ +--------+ +----------+
Event Flow (EventBus + OTel)¶
SimulationEngine ActionEngine CollaborationEngine
| | |
| SIMULATION_STARTED | ACTION_STARTED | SESSION_CREATED
| SIMULATION_STEP | ACTION_COMPLETED | TURN_TAKEN
| SIMULATION_COMPLETED | ACTION_FAILED | SESSION_COMPLETED
v v v
+------------------------------------------------------------------+
| EventBus |
| Thread-safe pub/sub, 81 event types, filtering, replay |
+-----+----+----+----+--------------------------------------------+
| | | |
v v v v
Global Filtered EventSubscriberPlugin OTel Exporter
Subs Subs (10th plugin type) (optional)
|
v
P1/FlowX event capture
(OTel integration)
Object Model Bridge Flow¶
+--------------------+ +--------------------+
| FCC Object Model | | External Models |
| | | |
| RepositoryProtocol |<------| CTO Repository[T] |
| ModelFacade ABC |<------| CTO CTOModel |
| VocabularyMapping |<------| CTO CrossVocab |
| EvolutionStage | | |
| DomainEntity | | (Future: AOME, |
| | | CONSTEL, etc.) |
+--------+-----------+ +--------------------+
|
| Adapter layer (cto_bridge.py)
|
+----+----+
| |
v v
CTOFacade CTORepository
Adapter Adapter
7. Plugin Dependency Triangle¶
The three plugin packages -- fcc-paom-plugin, fcc-aome-plugin, and
fcc-constel-plugin -- form a cyclic dependency triangle where each
provides a distinct capability to the next:
fcc-paom-plugin
/ (provides \
/ LLM routing) \
v \
fcc-aome-plugin provides_metadata
(provides taxonomy) |
\ |
\ |
v /
fcc-constel-plugin ----+
Dependency Types¶
| Source | Target | Type | Description |
|---|---|---|---|
fcc-paom-plugin |
fcc-aome-plugin |
provides_llm |
PAOM provides LLM routing for AOME privacy analysis |
fcc-aome-plugin |
fcc-constel-plugin |
provides_taxonomy |
AOME provides privacy taxonomy for CONSTEL metadata |
fcc-constel-plugin |
fcc-paom-plugin |
provides_metadata |
CONSTEL provides metadata for PAOM persona engineering |
These dependencies are declared in
src/fcc/data/ecosystem/plugin_dependencies.yaml.
Cross-Plugin Persona Interactions¶
Personas from different plugins interact across plugin boundaries. These
interactions are declared in
src/fcc/data/ecosystem/plugin_interactions.yaml:
| Source Persona | Source Plugin | Target Persona | Target Plugin | Type | Description |
|---|---|---|---|---|---|
| SNTR | fcc-paom-plugin | BDA | fcc-aome-plugin | downstream | SENTINEL Router sends bias-sensitive tasks to BDA |
| TXC | fcc-aome-plugin | KGA | fcc-constel-plugin | downstream | Taxonomy Curator feeds privacy vocabulary to KG Architect |
| UCR | fcc-constel-plugin | PDR | fcc-paom-plugin | downstream | UCM Registry Curator provides metadata for Persona Designer |
| DEO | fcc-aome-plugin | SNTN | fcc-paom-plugin | downstream | Data Ethics Officer escalates violations via SENTINEL Notifier |
| KGA | fcc-constel-plugin | INA | fcc-paom-plugin | peer | KG Architect and Inheritance Analyst share hierarchy data |
| PRA | fcc-aome-plugin | NSM | fcc-constel-plugin | peer | Privacy Risk Assessor coordinates with Namespace Manager |
CrossPluginOrchestrator¶
The CrossPluginOrchestrator class (src/fcc/plugins/orchestration.py)
manages the dependency triangle and provides runtime querying:
from fcc.plugins.orchestration import (
CrossPluginOrchestrator,
PluginDependency,
PluginInteraction,
PluginHealthStatus,
)
# Initialize with dependencies and interactions
orchestrator = CrossPluginOrchestrator(
dependencies=[
PluginDependency(
source_plugin="fcc-paom-plugin",
target_plugin="fcc-aome-plugin",
dependency_type="provides_llm",
description="PAOM provides LLM routing for AOME privacy analysis",
),
PluginDependency(
source_plugin="fcc-aome-plugin",
target_plugin="fcc-constel-plugin",
dependency_type="provides_taxonomy",
description="AOME provides privacy taxonomy for CONSTEL metadata",
),
PluginDependency(
source_plugin="fcc-constel-plugin",
target_plugin="fcc-paom-plugin",
dependency_type="provides_metadata",
description="CONSTEL provides metadata for PAOM persona engineering",
),
],
interactions=[
PluginInteraction(
source_persona="SNTR",
source_plugin="fcc-paom-plugin",
target_persona="BDA",
target_plugin="fcc-aome-plugin",
interaction_type="downstream",
description="SENTINEL Router sends bias-sensitive tasks to BDA",
),
# ... additional interactions
],
)
Querying Dependencies¶
# Forward dependencies: what does PAOM depend on?
paom_deps = orchestrator.resolve_dependencies("fcc-paom-plugin")
for dep in paom_deps:
print(f" -> {dep.target_plugin} ({dep.dependency_type})")
# Output: -> fcc-aome-plugin (provides_llm)
# Reverse dependencies: who depends on AOME?
aome_reverse = orchestrator.reverse_dependencies("fcc-aome-plugin")
for dep in aome_reverse:
print(f" <- {dep.source_plugin} ({dep.dependency_type})")
# Output: <- fcc-paom-plugin (provides_llm)
Querying Interactions¶
# Get the full interaction matrix (grouped by source plugin)
matrix = orchestrator.get_interaction_matrix()
for plugin, interactions in matrix.items():
print(f"\n{plugin}:")
for i in interactions:
print(f" {i.source_persona} -> {i.target_persona} "
f"({i.interaction_type})")
# Get all interactions for a specific persona
sntr_interactions = orchestrator.get_interactions_for_persona("SNTR")
for i in sntr_interactions:
print(f" {i.source_persona} -> {i.target_persona}: {i.description}")
Ecosystem Health Monitoring¶
# Check health across all plugins
statuses = [
PluginHealthStatus("fcc-paom-plugin", healthy=True, persona_count=13),
PluginHealthStatus("fcc-aome-plugin", healthy=True, persona_count=5),
PluginHealthStatus("fcc-constel-plugin", healthy=False, persona_count=5,
error="Missing TMF data file"),
]
report = orchestrator.check_health(statuses)
print(f"Total: {report.total_plugins}, "
f"Healthy: {report.healthy_plugins}, "
f"Unhealthy: {report.unhealthy_plugins}, "
f"Personas: {report.total_personas}")
# Output: Total: 3, Healthy: 2, Unhealthy: 1, Personas: 23
Validating Cross-Plugin Interactions¶
errors = orchestrator.validate_interactions()
if errors:
for err in errors:
print(f" Warning: {err}")
else:
print("All cross-plugin interactions are valid.")
8. Object Model Evolution Stages¶
When integrating external object models, use the evolution assessment to gauge maturity and plan integration depth:
| Stage | Score Threshold | Characteristics | Integration Strategy |
|---|---|---|---|
| Foundational | 0.00 | Flat entities, ad-hoc naming | Adapter only, manual mapping |
| Structured | 0.40 | Normalized schema, explicit relationships | Full adapter + VocabularyMapping |
| Semantic | 0.65 | Cross-vocabulary mappings, typed associations | Federated queries via ModelFacade |
| Federated | 0.85 | Multi-model federation, automated discovery | Full A2A + MCP integration |
from fcc.objectmodel.evolution import (
ObjectModelAssessment,
DimensionScore,
)
assessment = ObjectModelAssessment(
model_name="CTO",
dimensions=(
DimensionScore("schema_normalization", 0.90, "Fully normalized"),
DimensionScore("relationship_typing", 0.85, "Typed associations"),
DimensionScore("cross_vocabulary", 0.80, "CrossVocab mappings"),
DimensionScore("governance", 0.70, "Partial governance"),
),
)
print(f"Stage: {assessment.stage.value}") # "semantic"
print(f"Score: {assessment.aggregate_score}") # 0.8125
print(f"Gap to next: {assessment.gap_to_next()}") # 0.0375
print(f"Next stage: {assessment.next_stage().value}") # "federated"
# Identify weakest areas for improvement
for dim in assessment.weakest_dimensions(2):
print(f" Improve: {dim.name} (score: {dim.score})")
9. Quick Reference¶
Key Files¶
| File | Purpose |
|---|---|
src/fcc/objectmodel/base.py |
Protocol definitions (DomainEntity, RepositoryProtocol) |
src/fcc/objectmodel/facade.py |
ModelFacade ABC (stats, search, get_full) |
src/fcc/objectmodel/mapping.py |
VocabularyMapping, MappingStore protocol |
src/fcc/objectmodel/evolution.py |
EvolutionStage, ObjectModelAssessment |
src/fcc/objectmodel/cto_bridge.py |
CTO adapter (canonical integration pattern) |
src/fcc/objectmodel/examples.py |
ExampleRepository, ExampleFacade for testing |
src/fcc/plugins/orchestration.py |
CrossPluginOrchestrator, dependency/interaction models |
src/fcc/messaging/events.py |
Event, EventType (25 types) |
src/fcc/messaging/bus.py |
EventBus (thread-safe pub/sub) |
src/fcc/personas/cross_reference.py |
CrossReferenceMatrix (persona interaction map) |
src/fcc/scaffold/cli.py |
CLI commands (maps to MCP tools) |
src/fcc/data/ecosystem/project_registry.yaml |
Canonical ecosystem project registry |
src/fcc/data/ecosystem/plugin_dependencies.yaml |
Plugin dependency triangle |
src/fcc/data/ecosystem/plugin_interactions.yaml |
Cross-plugin persona interactions |
Integration Checklist¶
When adding a new ecosystem project integration:
- Define protocols in
src/fcc/objectmodel/base.py(or reuse existing ones) - Create frozen dataclasses for any new data transfer objects
- Write an adapter module following the pattern in
cto_bridge.py - Add a sentinel function (
project_available()) for the optional import - Register in ecosystem by adding an entry to
project_registry.yaml - Declare plugin dependencies in
plugin_dependencies.yamlif applicable - Map persona interactions in
plugin_interactions.yamlfor cross-plugin flows - Write tests that work both with and without the external package installed
- Assess evolution stage using
ObjectModelAssessmentto determine integration depth - Document A2A/MCP mappings if the project exposes agent or tool interfaces