API Reference¶

Desktop Agent¶

fdc3.desktop_agent.server ¶

FDC3 Desktop Agent server package.

This package provides the FastAPI application plus the WebSocket entrypoint that implements WCP/DACP messaging.

Most users will interact with:

create_app to embed the agent in another Python process.
app for the default application instance.

Submodules

app_factory: FastAPI application factory.
websocket: WebSocket endpoint handler.
routes: HTTP route handlers.
lifespan: Application lifecycle management (startup/shutdown).

AgentClientConnectionManager ¶

AgentClientConnectionManager()

Manages WebSocket connections for agent UI clients.

close_all `async` ¶

close_all()

Close all active agent-client WebSocket connections.

get_active_connection_count ¶

get_active_connection_count() -> int

Get count of active agent-client WebSocket connections.

BridgeClient ¶

BridgeClient(settings: BridgeConnectionSettings, *, implementation_metadata_factory: ImplementationMetadataFactory, channels_state_factory: ChannelsStateFactory, request_handler: RequestHandler, connected_agents_update_handler: Optional[Callable[[BridgeConnectedAgentsUpdatePayload], Awaitable[None]]] = None, connect_func: Optional[ConnectFunc] = None)

Desktop Agent Bridge client implementing BCP/BMP.

This is intentionally minimal: it supports - discovery + connect over the recommended port range (4475-4575) - BCP: hello -> handshake -> connectedAgentsUpdate - BMP: send request + await response by requestUuid - receive bridge-forwarded requests and respond via an injected handler

The bridge is expected to validate/augment meta.source.desktopAgent.

assigned_name `property` ¶

assigned_name: Optional[str]

Name assigned by the bridge during the handshake.

is_connected `property` ¶

is_connected: bool

Whether the bridge websocket is connected and assigned.

has_connected_agent ¶

has_connected_agent(name: str) -> bool

Return True if the named desktop agent is currently connected.

start `async` ¶

start() -> None

Start the bridge client loop if not already running.

stop `async` ¶

stop() -> None

Stop the bridge client and close resources.

Raises:

Type	Description
`Exception`	Re-raises errors from background tasks when shutting down.

send_agent_request `async` ¶

send_agent_request(*, request_type: str, payload: dict[str, Any], source: AppIdentifierLike, destination: Optional[AppIdentifierLike] = None, timeout: Optional[float] = None) -> BridgeResponseDict

Send an agentRequest message and await the (bridge-collated) response.

send_request_no_wait `async` ¶

send_request_no_wait(*, request_type: str, payload: dict[str, Any], source: AppIdentifierLike, destination: Optional[AppIdentifierLike] = None) -> str

Send a request message that does not generate a response.

Used for BMP 'fire and forget' messages such as broadcastRequest. Returns the generated requestUuid.

AccessControlHandler ¶

AccessControlHandler(access_control_manager: AccessControlManager, allowed_origins: List[str])

Handles WebSocket access control validation

validate_connection `async` ¶

validate_connection(websocket: WebSocket, headers: Mapping[str, str]) -> bool

Validate WebSocket connection based on access control policy.

Returns True if connection is allowed, False otherwise. Closes the WebSocket connection if access is denied.

WCPHandler ¶

WCPHandler(storage: Storage, *, bridge_enabled: bool = False)

Handles WCP (Web Connection Protocol) messages

handle_message `async` ¶

handle_message(message: Dict[str, Any], session_id: str, wcp_sessions: WcpSessions, sender: MessageSender) -> Optional[str]

Handle WCP message and return next phase if transition occurs.

Returns:

Type	Description
`Optional[str]`	"dacp" if transitioning to DACP phase
`Optional[str]`	None if staying in WCP phase

DACPHandler ¶

DACPHandler(storage: Storage, launcher: ProcessLauncher, connection_manager: WebSocketConnectionManager, web_launcher: Optional[WebEndpointLauncher] = None, core=None)

Bases: AppHandlersMixin, ChannelHandlersMixin, EventHandlersMixin, IntentHandlersMixin

Handles DACP (Desktop Agent Communication Protocol) messages

handle_message `async` ¶

handle_message(message: Dict[str, Any], session_id: str, wcp_sessions: WcpSessions, sender: MessageSender)

Handle DACP message with centralized Pydantic validation.

WebSocketConnectionManager ¶

WebSocketConnectionManager()

Manages WebSocket connections for app instances

add_connection ¶

add_connection(instance_uuid: str, websocket: WebSocket)

Add a WebSocket connection for an instance

remove_connection ¶

remove_connection(instance_uuid: str)

Remove a WebSocket connection for an instance

send_to_instance `async` ¶

send_to_instance(instance_uuid: str, message: str)

Send a message to a specific instance

get_connected_instances ¶

get_connected_instances()

Get list of connected instance UUIDs

get_connection_count ¶

get_connection_count() -> int

Get count of connected instances.

close_all `async` ¶

close_all()

Close all active WebSocket connections.

create_app ¶

create_app(config: Optional[DesktopAgentConfig] = None) -> FastAPI

Create and configure the FDC3 Desktop Agent FastAPI application.

This factory function allows embedding the agent in another Python project with full control over configuration.

Parameters:

Name	Type	Description	Default
`config`	`Optional[DesktopAgentConfig]`	Optional configuration. If None, uses environment variables and defaults.	`None`

Returns:

Type	Description
`FastAPI`	A configured FastAPI application ready to be run with uvicorn or
`FastAPI`	mounted into a larger ASGI application.

Example

Standalone usage¶

app = create_app()

With custom config¶

config = DesktopAgentConfig(host="0.0.0.0", port=9000, db_path=":memory:") app = create_app(config)

Mount in larger app¶

main_app = FastAPI() main_app.mount("/fdc3", create_app())

fdc3.desktop_agent.cli ¶

Command-line stub to launch the FDC3 Desktop Agent.

Invokes uvicorn (or uv run uvicorn) in a subprocess and forwards Ctrl-C / SIGTERM so shutdown logs complete before the shell prompt returns.

main ¶

main(argv: List[str] | None = None) -> int

Launch the desktop agent or manage the app directory.

If no arguments are provided, starts uvicorn in dev mode with reload.

fdc3.desktop_agent.config ¶

Configuration for the FDC3 Desktop Agent.

This module provides the DesktopAgentConfig dataclass used to configure the agent when embedding it in another Python application.

DesktopAgentConfig `dataclass` ¶

DesktopAgentConfig(host: str = (lambda: getenv('FDC3_HOST', 'localhost'))(), port: int = (lambda: int(getenv('FDC3_PORT', '8000')))(), db_path: str = (lambda: getenv('FDC3_DB_PATH', 'fdc3_agent.db'))(), log_level: str = (lambda: getenv('FDC3_LOG_LEVEL', 'INFO'))(), allowed_origins: List[str] = _default_allowed_origins(), storage: Optional['Storage'] = None, launcher: Optional['ProcessLauncher'] = None, web_launcher: Optional['WebEndpointLauncher'] = None, distributed_adapter: Optional['DistributedLogAdapter'] = None, plugins: List['IntentHandlerPlugin'] = list(), auto_discover_plugins: bool = (lambda: lower() in ('true', '1', 'yes'))(), agent_url: Optional[str] = None, bridge_enabled: bool = (lambda: lower() in ('true', '1', 'yes'))(), bridge_host: str = (lambda: getenv('FDC3_BRIDGE_HOST', '127.0.0.1'))(), bridge_port_start: int = (lambda: int(getenv('FDC3_BRIDGE_PORT_START', '4475')))(), bridge_port_end: int = (lambda: int(getenv('FDC3_BRIDGE_PORT_END', '4575')))(), bridge_requested_name: str = (lambda: getenv('FDC3_BRIDGE_REQUESTED_NAME', gethostname()))(), bridge_connect_retry_seconds: float = (lambda: float(getenv('FDC3_BRIDGE_CONNECT_RETRY_SECONDS', '5')))(), bridge_request_timeout_seconds: float = (lambda: float(getenv('FDC3_BRIDGE_REQUEST_TIMEOUT_SECONDS', '3')))())

Configuration for the FDC3 Desktop Agent.

All fields have sensible defaults, allowing you to create a minimal configuration with just DesktopAgentConfig().

When embedding the agent, you can override any field::

config = DesktopAgentConfig(
    host="0.0.0.0",
    port=9000,
    db_path=":memory:",
)
app = create_app(config)

You can also inject custom implementations of storage, launcher, or distributed adapter::

config = DesktopAgentConfig(
    storage=MyCustomStorage(),
    launcher=MyCustomLauncher(),
)

computed_agent_url `property` ¶

computed_agent_url: str

Return the agent WebSocket URL, computing from host/port if not set.

templates_dir `property` ¶

templates_dir: Path

Return the path to the templates directory.

Examples¶

Start the agent quickly with Docker:

docker-compose up

See getting-started.md for the full walkthrough.

Client¶

fdc3.client.client ¶

Async client for connecting external intent handlers to the desktop agent.

The primary entry point is FDC3Client, which implements the WebSocket Connection Protocol (WCP) handshake and provides helpers for registering an external handler, subscribing to context/intent notifications, and sending results.

Typical usage:

```python
import asyncio

from fdc3.client.client import FDC3Client


async def main() -> None:
    async with FDC3Client("ws://localhost:8000/ws", handler_id="my-handler") as c:
        await c.register_handler("my-handler", intents=["ViewChart"])

        async def on_intent(msg):
            # msg is a validated Pydantic model for known message types.
            await c.send_intent_result(msg.meta["requestUuid"], result={"type": "fdc3.nothing"})

        c.forwarded_intent_handlers.add(on_intent)
        await c.run_forever()


asyncio.run(main())
```

DACPRequest ¶

Bases: Protocol

Protocol for DACP request models with meta attribute.

FDC3Client ¶

FDC3Client(agent_url: str, handler_id: str = 'external-handler', *, ping_interval: float = 20.0)

Client for external intent handlers to connect to the FDC3 desktop agent.

The client is designed for external intent handler processes that need to:

establish a WebSocket connection to an agent;
complete the WCP handshake;
register/unregister an external handler and supported intents;
receive forwarded intents and broadcasts via EventEmitter.

Notes

This is an asyncio-based client.
Message handlers registered on the public emitters receive validated Pydantic models for known message types.

send_dacp_request `async` ¶

send_dacp_request(request: DACPRequest, timeout: float = 5.0) -> Any

Send a DACP request model and wait for the correlated response.

Parameters:

Name	Type	Description	Default
`request`	`DACPRequest`	A Pydantic DACP request model (e.g., generated from the `fdc3.models.dacp` module) that includes a `meta.requestUuid`.	required
`timeout`	`float`	Seconds to wait for the correlated response.	`5.0`

Returns:

Type	Description
`Any`	The parsed response payload (model or raw dict) returned by the agent.

Raises:

Type	Description
`TimeoutError`	If a response is not received within `timeout`.
`RuntimeError`	If the client is not connected or handshake not complete.

connect `async` ¶

connect() -> None

Connect to the agent and complete the WCP handshake.

After this returns successfully, wait_for_handshake should be satisfied and request/response helper methods (e.g. register_handler) can be used.

Raises:

Type	Description
`Exception`	If the websocket connection or handshake fails.

close `async` ¶

close() -> None

Close the websocket connection and stop background tasks.

wait_for_handshake `async` ¶

wait_for_handshake(timeout: float = 10.0) -> bool

Wait for WCP handshake to complete.

Returns:

Type	Description
`bool`	True if handshake succeeded, False if timed out.

register_handler `async` ¶

register_handler(handler_id: str, intents: List[str], *, priority: int = 0, metadata: Optional[Dict[str, Any]] = None, timeout: float = 5.0) -> str

Register an external handler and return the agent-assigned handler UUID.

Parameters:

Name	Type	Description	Default
`handler_id`	`str`	A stable identifier for this handler.	required
`intents`	`List[str]`	Intent names the handler can service.	required
`priority`	`int`	Higher values may win in resolver selection (agent-dependent).	`0`
`metadata`	`Optional[Dict[str, Any]]`	Arbitrary handler metadata for discovery/UI.	`None`
`timeout`	`float`	Seconds to wait for the correlated response.	`5.0`

Returns:

Type	Description
`str`	The agent-assigned handler UUID.

add_context_listener `async` ¶

add_context_listener(context_type: Optional[str] = None, timeout: float = 5.0) -> str

Register a context listener and return the listener UUID.

Parameters:

Name	Type	Description	Default
`context_type`	`Optional[str]`	If provided, only contexts of this FDC3 type are delivered.	`None`
`timeout`	`float`	Seconds to wait for the correlated response.	`5.0`

Returns:

Type	Description
`str`	Listener UUID that can be passed to `remove_context_listener`.

remove_context_listener `async` ¶

remove_context_listener(listener_uuid: str, timeout: float = 5.0) -> None

Unsubscribe a previously-registered context listener.

add_intent_listener `async` ¶

add_intent_listener(intent: str, timeout: float = 5.0) -> str

Register an intent listener and return its listener UUID.

Parameters:

Name	Type	Description	Default
`intent`	`str`	The intent name to listen for (e.g., "ViewChart").	required
`timeout`	`float`	Seconds to wait for the agent's response.	`5.0`

Returns:

Type	Description
`str`	Listener UUID that can be passed to `remove_intent_listener`.

remove_intent_listener `async` ¶

remove_intent_listener(listener_uuid: str, timeout: float = 5.0) -> None

Unsubscribe a previously-registered intent listener.

Parameters:

Name	Type	Description	Default
`listener_uuid`	`str`	The UUID returned from `add_intent_listener`.	required
`timeout`	`float`	Seconds to wait for the agent's response.	`5.0`

unregister_handler `async` ¶

unregister_handler(handler_uuid: str, timeout: float = 5.0) -> None

Unregister a handler by its UUID.

Parameters:

Name	Type	Description	Default
`handler_uuid`	`str`	The UUID returned from `register_handler`.	required
`timeout`	`float`	Seconds to wait for the agent's response.	`5.0`

send_intent_result `async` ¶

send_intent_result(request_uuid: str, *, result: Optional[Dict[str, Any]] = None, error: Optional[str] = None) -> None

Send a result for a previously forwarded intent.

Parameters:

Name	Type	Description	Default
`request_uuid`	`str`	The request UUID from the forwarded intent message meta.	required
`result`	`Optional[Dict[str, Any]]`	Optional result payload.	`None`
`error`	`Optional[str]`	Optional error string.	`None`

Notes

Many handler processes call this from a callback registered via :attr:forwarded_intent_handlers.

emit_channel_event `async` ¶

emit_channel_event(event_type: str, channel_id: str, *, instance_uuid: Optional[str] = None, context: Optional[Dict[str, Any]] = None) -> None

Emit a dev-only channel event via the server GraphQL emitChannelEvent mutation.

This is intended for examples and demos only.

create_user_channel `async` ¶

create_user_channel(channel_id: str, *, display_metadata: Optional[DisplayMetadata] = None) -> None

Create a user channel via the agent GraphQL API.

Parameters:

Name	Type	Description	Default
`channel_id`	`str`	The desired channel identifier (e.g., "demo" or "user:demo").	required
`display_metadata`	`Optional[DisplayMetadata]`	Optional `DisplayMetadata` providing a human name and color for the channel.	`None`

Raises:

Type	Description
`HTTPError`	If the GraphQL request fails.

join_user_channel `async` ¶

join_user_channel(channel_id: str, *, auto_create: bool = False) -> JoinUserChannelResponse

Join a user channel, optionally auto-creating it if missing.

Parameters:

Name	Type	Description	Default
`channel_id`	`str`	The channel identifier (e.g., "demo" or "user:demo").	required
`auto_create`	`bool`	If True and the channel doesn't exist, create it first.	`False`

Returns:

Type	Description
`JoinUserChannelResponse`	The JoinUserChannelResponse containing the joined channel details.

Raises:

Type	Description
`Exception`	If the channel doesn't exist and auto_create is False.

leave_current_channel `async` ¶

leave_current_channel() -> None

Leave the currently joined channel.

This sends a DACP leaveCurrentChannel request to the agent.

Raises:

Type	Description
`Exception`	If the agent returns an error or the request fails.

create_private_channel `async` ¶

create_private_channel(display_name: Optional[str] = None) -> CreatePrivateChannelResponse

Create a private channel and return the response.

Parameters:

Name	Type	Description	Default
`display_name`	`Optional[str]`	Optional human-readable name for the channel.	`None`

Returns:

Type	Description
`CreatePrivateChannelResponse`	The CreatePrivateChannelResponse containing the new channel's metadata.

create_private_channel_invite `async` ¶

create_private_channel_invite(channel_id: str, instance_id: Optional[str] = None) -> CreatePrivateChannelInvitationResponse

Create a private channel invitation.

Parameters:

Name	Type	Description	Default
`channel_id`	`str`	The private channel to create an invitation for.	required
`instance_id`	`Optional[str]`	Optional target instance to restrict the invitation to.	`None`

Returns:

Type	Description
`CreatePrivateChannelInvitationResponse`	The CreatePrivateChannelInvitationResponse containing the invitation token.

join_private_channel `async` ¶

join_private_channel(channel_id: str, token: str) -> JoinPrivateChannelResponse

Join a private channel using an invitation token.

Parameters:

Name	Type	Description	Default
`channel_id`	`str`	The private channel to join.	required
`token`	`str`	The invitation token received from the channel creator.	required

Returns:

Type	Description
`JoinPrivateChannelResponse`	The JoinPrivateChannelResponse confirming the join.

leave_private_channel `async` ¶

leave_private_channel(channel_id: str) -> None

Leave a private channel.

Parameters:

Name	Type	Description	Default
`channel_id`	`str`	The private channel to leave.	required

add_private_channel_event_listener `async` ¶

add_private_channel_event_listener(channel_id: str, *, event_type: Optional[PrivateChannelEventListenerTypes] = None) -> PrivateChannelAddEventListenerResponse

Subscribe to private channel events for a channel.

Parameters:

Name	Type	Description	Default
`channel_id`	`str`	The private channel to subscribe to.	required
`event_type`	`Optional[PrivateChannelEventListenerTypes]`	Optionally filter to a specific event type.	`None`

Returns:

Type	Description
`PrivateChannelAddEventListenerResponse`	The PrivateChannelAddEventListenerResponse containing listener details.

build_message `async` ¶

build_message(text: str) -> MessageContext

Build a minimal fdc3.message payload for a chat message.

Parameters:

Name	Type	Description	Default
`text`	`str`	The plain text content of the message.	required

Returns:

Type	Description
`MessageContext`	A `MessageContext` object ready for use in chat operations.

get_chat_room `async` ¶

get_chat_room(channel_id: str, *, provider_name: Optional[str] = None, auto_create: bool = False) -> ChatRoomContext

Return a fdc3.chat.room object for a given channel id.

Parameters:

Name	Type	Description	Default
`channel_id`	`str`	The channel identifier to build a chat room for.	required
`provider_name`	`Optional[str]`	Optional provider name to include in the room.	`None`
`auto_create`	`bool`	If True, attempt to create a user channel on the agent (best-effort) before returning the room object.	`False`

Returns:

Type	Description
`ChatRoomContext`	A `ChatRoomContext` representing the chat room for the given
`ChatRoomContext`	`channel_id`.

send_chat_message `async` ¶

send_chat_message(text: str, channel_id: str, *, provider_name: Optional[str] = None, auto_create_room: bool = False) -> None

Send a fdc3.chat.message to the agent by broadcasting the appropriate context object.

This constructs the chatRoom and message objects and calls broadcast with the resulting context.

broadcast `async` ¶

broadcast(context: Any) -> None

Send a DACP broadcast request to the agent to broadcast context.

This will cause the agent to deliver the context to the channel the sending instance is currently joined to.

Parameters:

Name	Type	Description	Default
`context`	`Any`	An FDC3 context object (dict or Pydantic model) to broadcast.	required

Example

await client.broadcast({"type": "fdc3.instrument", "id": {"ticker": "AAPL"}})

run_forever `async` ¶

run_forever() -> None

Block until the connection closes, then clean up.

This is a convenience for long-running external handler processes.

Examples¶

Register an external handler and run the client:

import asyncio

from fdc3.client.client import FDC3Client


async def main() -> None:
    async with FDC3Client("ws://localhost:8000/ws", handler_id="my-handler") as c:
        await c.register_handler("my-handler", intents=["ViewChart"])
        await c.run_forever()


asyncio.run(main())

See getting-started.md for a minimal demo client.

Models¶

fdc3.models.identifiers ¶

Identifier and small domain types.

This module contains compact Pydantic models used across the codebase and was previously re-exported from the generated fdc3.desktop_agent.api. Keeping the definitions here avoids import cycles and provides a stable surface for other modules.

API Reference¶

Desktop Agent¶

fdc3.desktop_agent.server ¶

AgentClientConnectionManager ¶

close_all async ¶

get_active_connection_count ¶

BridgeClient ¶

assigned_name property ¶

is_connected property ¶

has_connected_agent ¶

start async ¶

stop async ¶

send_agent_request async ¶

send_request_no_wait async ¶

AccessControlHandler ¶

validate_connection async ¶

WCPHandler ¶

handle_message async ¶

DACPHandler ¶

handle_message async ¶

WebSocketConnectionManager ¶

add_connection ¶

remove_connection ¶

send_to_instance async ¶

get_connected_instances ¶

get_connection_count ¶

close_all async ¶

create_app ¶

Standalone usage¶

With custom config¶

Mount in larger app¶

fdc3.desktop_agent.cli ¶

main ¶

fdc3.desktop_agent.config ¶

DesktopAgentConfig dataclass ¶

computed_agent_url property ¶

templates_dir property ¶

Examples¶

Client¶

fdc3.client.client ¶

DACPRequest ¶

FDC3Client ¶

send_dacp_request async ¶

connect async ¶

close async ¶

wait_for_handshake async ¶

register_handler async ¶

add_context_listener async ¶

remove_context_listener async ¶

add_intent_listener async ¶

remove_intent_listener async ¶

unregister_handler async ¶

send_intent_result async ¶

emit_channel_event async ¶

create_user_channel async ¶

join_user_channel async ¶

leave_current_channel async ¶

create_private_channel async ¶

create_private_channel_invite async ¶

join_private_channel async ¶

leave_private_channel async ¶

add_private_channel_event_listener async ¶

build_message async ¶

get_chat_room async ¶

send_chat_message async ¶

broadcast async ¶

run_forever async ¶

Examples¶

Models¶

fdc3.models.identifiers ¶

close_all `async` ¶

assigned_name `property` ¶

is_connected `property` ¶

start `async` ¶

stop `async` ¶

send_agent_request `async` ¶

send_request_no_wait `async` ¶

validate_connection `async` ¶

handle_message `async` ¶

handle_message `async` ¶

send_to_instance `async` ¶

close_all `async` ¶

DesktopAgentConfig `dataclass` ¶

computed_agent_url `property` ¶

templates_dir `property` ¶

send_dacp_request `async` ¶

connect `async` ¶

close `async` ¶

wait_for_handshake `async` ¶

register_handler `async` ¶

add_context_listener `async` ¶

remove_context_listener `async` ¶

add_intent_listener `async` ¶

remove_intent_listener `async` ¶

unregister_handler `async` ¶

send_intent_result `async` ¶

emit_channel_event `async` ¶

create_user_channel `async` ¶

join_user_channel `async` ¶

leave_current_channel `async` ¶

create_private_channel `async` ¶

create_private_channel_invite `async` ¶

join_private_channel `async` ¶

leave_private_channel `async` ¶

add_private_channel_event_listener `async` ¶

build_message `async` ¶

get_chat_room `async` ¶

send_chat_message `async` ¶

broadcast `async` ¶

run_forever `async` ¶