API Reference: Protocols Module¶

This document covers the FCC protocol integration API surface, including A2A agent card generation, MCP server with tools/resources/prompts, cross-protocol event routing via the Protocol Bridge, WebSocket/SSE streaming, and multi-stream event management.

sequenceDiagram
    participant Client as External Agent
    participant Bridge as ProtocolBridge
    participant A2A as A2A CardBuilder
    participant MCP as FccMcpServer
    participant Bus as EventBus

    Client->>Bridge: route("a2a", task)
    Bridge->>Bus: publish(A2A_TASK_RECEIVED)
    Bridge->>A2A: handle task
    A2A-->>Bridge: AgentCard response
    Bridge-->>Client: result

    Client->>Bridge: route("mcp", tool_call)
    Bridge->>Bus: publish(MCP_TOOL_CALLED)
    Bridge->>MCP: call_tool()
    MCP-->>Bridge: tool result
    Bridge-->>Client: result

Overview¶

The protocol integration layer exposes FCC capabilities through three industry-standard protocols:

Protocol	Module Area	Purpose
A2A (Agent-to-Agent)	Agent cards, skill definitions	Cross-agent capability discovery
MCP (Model Context Protocol)	Tools, resources, prompts	LLM host integration
EventBus Bridge	Event streaming, WebSocket	Real-time event delivery

A2A Agent Cards¶

Data Models¶

All models are frozen dataclasses in fcc.protocols.a2a.models:

AgentCard -- Full agent identity card with id, name, description, version, url, endpoint, capabilities, skills, and metadata. Includes to_dict(), from_dict(), to_json(), and validate_against_schema() methods.
SkillDefinition -- A discrete skill with id, name, description, input_schema, output_schema, and tags. Mapped from WorkflowAction instances.
AgentCapability -- A named capability with name, description, and tags. Derived from R.I.S.C.E.A.R. archetypes and role skills.
AgentEndpoint -- Network endpoint with url and protocol (default "a2a/v1").

from fcc.protocols.a2a.models import AgentCard, SkillDefinition, AgentCapability

# Deserialize from JSON
card = AgentCard.from_dict(json_data)

# Validate against JSON Schema
errors = card.validate_against_schema()
if not errors:
    print("Card is valid")

# Serialize to JSON
json_str = card.to_json(indent=2)

CardBuilder¶

fcc.protocols.a2a.card_builder.CardBuilder translates persona specs into A2A agent cards. It maps R.I.S.C.E.A.R. fields as follows:

R.I.S.C.E.A.R. Field	Agent Card Field
`persona.name`	`card.name`
`persona.id`	`card.id`
`persona.riscear.role`	`card.description`
`persona.riscear.archetype`	Primary capability
`persona.riscear.role_skills`	Additional capabilities
`WorkflowAction` instances	Skill definitions
`persona.category`, `persona.fcc_phase`	Metadata

from fcc.personas.registry import PersonaRegistry
from fcc.protocols.a2a.card_builder import CardBuilder
from fcc._resources import get_personas_dir

registry = PersonaRegistry.from_yaml_directory(get_personas_dir())
builder = CardBuilder(base_url="https://agents.example.com")

# Build a single card
card = builder.build_card(registry.get("RC"))

# Build cards for all personas
cards = builder.build_all_cards(registry)

# Write .well-known/agent.json
from pathlib import Path
builder.write_well_known(cards[0], Path("/tmp/output"))

AgentsMdGenerator¶

fcc.protocols.agents_md.AgentsMdGenerator produces a standard AGENTS.md manifest following the Google A2A convention.

from fcc.protocols.agents_md import AgentsMdGenerator

gen = AgentsMdGenerator.from_registry(registry, base_url="https://agents.example.com")
gen.write(Path("AGENTS.md"))
print(f"Generated {len(gen)} agent entries")

MCP Server¶

fcc.protocols.mcp.server.FccMcpServer serves tools, resources, and prompts via the MCP protocol. It manages three registries and dispatches calls to registered handlers, publishing lifecycle events to the EventBus.

Default Registries¶

Registry	Default Count	Class
Tools	14	`MCPToolRegistry.create_default()`
Resources	14	`MCPResourceRegistry.create_default()`
Prompts	6	`MCPPromptRegistry.create_default()`

Server Setup and Tool Execution¶

from fcc.messaging.bus import EventBus
from fcc.protocols.mcp.server import FccMcpServer

bus = EventBus()
server = FccMcpServer(event_bus=bus)

# Register a tool handler
def my_handler(persona_id: str) -> dict:
    return {"status": "ok", "persona_id": persona_id}

server.register_tool_handler("fcc_get_persona", my_handler)

# Call the tool -- publishes MCP events to EventBus
result = server.call_tool("fcc_get_persona", {"persona_id": "RC"})
# result: {"status": "ok", "result": {"status": "ok", "persona_id": "RC"}}

Resource Reading¶

# Read a resource
resource = server.read_resource("fcc://personas/RC")
# resource: {"status": "ok", "data": {...}, "mime_type": "application/json"}

Prompt Rendering¶

prompt = server.get_prompt("fcc_review_prompt", {"topic": "API design"})
# prompt: {"description": "...", "messages": [...]}

MCP Resource URIs¶

URI Pattern	Description
`fcc://personas`	Full persona registry
`fcc://personas/{id}`	Individual persona by ID
`fcc://workflows`	Available workflow graphs
`fcc://workflows/{name}`	Specific workflow graph
`fcc://scenarios`	Simulation scenarios
`fcc://constitutions/{persona_id}`	Constitution rules
`fcc://quality-gates`	Quality gate definitions
`fcc://cross-references`	Cross-reference matrix
`fcc://tags`	Tag registry
`fcc://events/types`	Event type catalog
`fcc://actions/{persona_id}`	Workflow actions
`fcc://plugins`	Plugin registry
`fcc://ecosystem`	Ecosystem metadata
`fcc://collaboration/sessions`	Active sessions

MCP Events¶

The server publishes the following events to the EventBus when configured:

Event Type	Trigger
`PROTOCOL_MCP_TOOL_CALLED`	Before tool execution
`PROTOCOL_MCP_TOOL_COMPLETED`	After successful tool execution
`PROTOCOL_MCP_TOOL_FAILED`	On tool execution error
`PROTOCOL_MCP_RESOURCE_READ`	On resource access

Protocol Bridge¶

fcc.protocols.bridge.ProtocolBridge acts as a central dispatcher that routes incoming protocol messages to registered handlers while publishing lifecycle events to the EventBus.

from fcc.messaging.bus import EventBus
from fcc.protocols.bridge import ProtocolBridge

bus = EventBus()

# Create with default A2A and MCP stub routes
bridge = ProtocolBridge.create_default(bus)

# Route a message
result = bridge.route("a2a", {"task_id": "t1", "skill": "review"})
# result: {"status": "accepted", "protocol": "a2a", "message_id": "t1"}

# Register a custom route
def my_protocol_handler(message):
    return {"status": "processed", "data": message}

bridge.register_route("custom", my_protocol_handler)

# Check routes
print(bridge.list_routes())    # ["a2a", "custom", "mcp"]
print(bridge.has_route("a2a"))  # True

Event Publishing¶

Protocol	Entry Event Type
`"a2a"`	`PROTOCOL_A2A_TASK_RECEIVED`
`"mcp"`	`PROTOCOL_MCP_TOOL_CALLED`

Unknown protocols or handler errors publish PROTOCOL_BRIDGE_ERROR.

WebSocket Bridge¶

fcc.protocols.ws_bridge.WebSocketBridge bridges EventBus events to browser clients over WebSocket connections. Requires the optional websockets package.

from fcc.protocols.ws_bridge import WebSocketBridge, ws_bridge_available

if ws_bridge_available():
    bridge = WebSocketBridge(host="localhost", port=8765)

    # Create an EventBus subscriber
    from fcc.messaging.bus import EventBus
    bus = EventBus()
    subscriber = bridge.create_event_bus_subscriber()
    bus.subscribe(subscriber)

    # Start the server (async)
    import asyncio
    asyncio.run(bridge.start())

Property	Type	Description
`host`	`str`	Hostname the server binds to
`port`	`int`	Port the server binds to
`client_count`	`int`	Number of connected clients
`is_running`	`bool`	Whether the server is active

Event Streaming Configuration¶

fcc.protocols.streaming provides frozen dataclass models for stream configuration and filtering.

StreamFilter¶

from fcc.protocols.streaming import StreamFilter

filter = StreamFilter(
    event_types=("SIMULATION_STARTED", "SIMULATION_COMPLETED"),
    sources=("simulation.engine",),
    correlation_id="run-001",
)

event_data = {"event_type": "SIMULATION_STARTED", "source": "simulation.engine"}
assert filter.matches(event_data)

EventStreamConfig¶

from fcc.protocols.streaming import EventStreamConfig, StreamFilter

config = EventStreamConfig(
    stream_id="sim-events",
    format="websocket",
    filter=StreamFilter(event_types=("TURN_COMPLETED",)),
    buffer_size=1000,
    heartbeat_interval_seconds=30.0,
    reconnect_delay_seconds=3.0,
)
serialized = config.to_dict()
restored = EventStreamConfig.from_dict(serialized)

Stream Manager¶

fcc.protocols.StreamManager manages multiple named event streams, optionally integrating with a VisualizationBridge for D3-ready payload transformation.

from fcc.protocols import StreamManager
from fcc.messaging.events import Event, EventType

manager = StreamManager()

# Create named streams
manager.create_stream("simulation", {"type": "sim"})
manager.create_stream("collaboration", {"type": "collab"})

# Push events
event = Event(event_type=EventType.SIMULATION_STARTED, source="engine")
manager.push_event("simulation", event)

# Broadcast to all streams
count = manager.push_to_all(event)

# Query state
print(manager.active_streams())       # ["collaboration", "simulation"]
print(manager.stream_count())         # 2
stats = manager.stream_stats("simulation")

Stat Key	Description
`stream_id`	Stream identifier
`event_count`	Number of events pushed
`d3_payload_count`	Number of D3 payloads generated
`has_viz_bridge`	Whether a visualization bridge is attached
`created_at`	ISO 8601 creation timestamp

Protocol Engineering Personas¶

ID	Name	Category	Responsibility
ASD	A2A Skill Designer	protocol_engineering	Agent card and skill specifications
MTA	MCP Tool Architect	protocol_engineering	MCP tool schemas and resource URIs
EBO	Event Bridge Orchestrator	protocol_engineering	EventBus to protocol bridging
WSM	WebSocket Stream Manager	protocol_engineering	Real-time event transport
PCA	Protocol Compliance Auditor	protocol_engineering	Protocol conformance verification
AMG	AGENTS.md Generator	protocol_engineering	Human-readable agent documentation

Import Paths Summary¶

Class	Import Path
`AgentCard`	`fcc.protocols.a2a.models`
`SkillDefinition`	`fcc.protocols.a2a.models`
`AgentCapability`	`fcc.protocols.a2a.models`
`AgentEndpoint`	`fcc.protocols.a2a.models`
`CardBuilder`	`fcc.protocols.a2a.card_builder`
`AgentsMdGenerator`	`fcc.protocols.agents_md`
`FccMcpServer`	`fcc.protocols.mcp.server`
`MCPToolRegistry`	`fcc.protocols.mcp.tools`
`MCPResourceRegistry`	`fcc.protocols.mcp.resources`
`MCPPromptRegistry`	`fcc.protocols.mcp.prompts`
`ProtocolBridge`	`fcc.protocols.bridge`
`WebSocketBridge`	`fcc.protocols.ws_bridge`
`StreamManager`	`fcc.protocols.stream_manager`
`EventStreamConfig`	`fcc.protocols.streaming`
`StreamFilter`	`fcc.protocols.streaming`