agenticraft_foundation.integration.mpst_bridge

MPST bridge adapter — maps MCP and A2A protocol interactions to multiparty session types for protocol-verified communication.

MPST integration adapter for protocol bridges.

This module provides session type verification for cross-protocol messaging.

Key Features: - Session type verification before message routing - Protocol session type mapping (MCP, A2A → MPST types) - Runtime session conformance monitoring - Well-formedness checking for protocol session types

ProtocolName

Bases: str, Enum

Supported protocol names.

SessionStatus

Bases: str, Enum

Status of a session.

ProtocolSessionType dataclass

Maps a protocol interaction to an MPST session type.

ATTRIBUTE	DESCRIPTION
protocol	The protocol name TYPE: ProtocolName
method	The protocol method (e.g., "tools/call", "tasks/send") TYPE: str
session_type	The corresponding MPST global type TYPE: SessionType
participants	Participant roles in the session TYPE: frozenset[ParticipantId]

SessionVerificationResult dataclass

Result of session type verification.

ATTRIBUTE	DESCRIPTION
is_valid	Whether the session conforms to its type TYPE: bool
session_type	The verified session type TYPE: SessionType | None
violations	List of conformance violations TYPE: list[str]
current_status	Current status of the session TYPE: SessionStatus
remaining_type	Remaining session type to complete TYPE: SessionType | None

valid(session_type, remaining=None) classmethod

Create a valid verification result.

invalid(violations, session_type=None) classmethod

Create an invalid verification result.

ActiveSession dataclass

Internal representation of an active session.

MPSTBridgeAdapter

Adapter integrating MPST session types with ProtocolBridge.

This adapter provides session type verification for cross-protocol message routing. It hooks into the ProtocolBridge's translation and routing pipeline to ensure session conformance.

Usage

adapter = MPSTBridgeAdapter()

Register session type for a protocol method

adapter.register_session_type(mcp_tool_call_session())

Verify message conforms to session type

result = adapter.verify_message(message, "tools/call", ProtocolName.MCP)

Start a monitored session

session_id = await adapter.start_session( session_type=mcp_tool_call_session().session_type, participants={"client": "agent-1", "server": "agent-2"}, )

Monitor message in session

await adapter.on_message(session_id, message)

__init__()

Initialize the adapter.

register_session_type(session_type)

Register a session type for a protocol method.

PARAMETER	DESCRIPTION
session_type	The session type to register TYPE: ProtocolSessionType

get_session_type(protocol, method)

Get the session type for a protocol method.

PARAMETER	DESCRIPTION
protocol	The protocol name TYPE: ProtocolName
method	The method name TYPE: str

RETURNS	DESCRIPTION
ProtocolSessionType | None	The session type, or None if not registered

verify_wellformedness(session_type)

Verify a session type is well-formed.

PARAMETER	DESCRIPTION
session_type	The global type to verify TYPE: SessionType

RETURNS	DESCRIPTION
SessionVerificationResult	Verification result

verify_message(message, method, protocol, participant=None)

Verify a message conforms to the expected session type.

PARAMETER	DESCRIPTION
message	The message to verify TYPE: dict[str, Any]
method	The protocol method TYPE: str
protocol	The protocol name TYPE: ProtocolName
participant	Optional participant to verify as TYPE: ParticipantId | None DEFAULT: None

RETURNS	DESCRIPTION
SessionVerificationResult	Verification result

start_session(session_type, participants) async

Start a new monitored session.

PARAMETER	DESCRIPTION
session_type	The global session type TYPE: SessionType
participants	Mapping of participant roles to agent IDs TYPE: dict[ParticipantId, str]

RETURNS	DESCRIPTION
str	Session ID for tracking

RAISES	DESCRIPTION
ValueError	If session type is not well-formed

on_message(session_id, message, sender, receiver) async

Process a message in an active session.

PARAMETER	DESCRIPTION
session_id	The session ID TYPE: str
message	The message being sent TYPE: dict[str, Any]
sender	The sending participant role TYPE: ParticipantId
receiver	The receiving participant role TYPE: ParticipantId

RETURNS	DESCRIPTION
SessionVerificationResult	Verification result

end_session(session_id) async

End a session and verify it completed properly.

PARAMETER	DESCRIPTION
session_id	The session ID TYPE: str

RETURNS	DESCRIPTION
SessionVerificationResult	Final verification result

get_active_sessions()

Get list of active session IDs.

get_session_status(session_id)

Get current status of all participants in a session.

PARAMETER	DESCRIPTION
session_id	The session ID TYPE: str

RETURNS	DESCRIPTION
dict[ParticipantId, SessionStatus] | None	Mapping of participant to status, or None if session not found

mcp_tool_call_session()

MCP tools/call session type: client → server : call. server → client : result.

mcp_resource_read_session()

MCP resources/read session type.

a2a_task_send_session()

A2A tasks/send session type with status updates.