use harn_serve::adapters::acp::{
ACP_SCHEMA_COMPATIBILITY, HARN_AGENT_EVENT_KINDS, HARN_AGENT_EVENT_METHOD,
HARN_CONTENT_EXTENSION_FIELDS, HARN_PROVIDER_CATALOG_METHOD, HARN_SESSION_UPDATE_EXTENSIONS,
HARN_TOOL_LIFECYCLE_EXTENSION_FIELDS,
};
use harn_serve::{A2A_PROTOCOL_VERSION, MCP_PROTOCOL_VERSION};
use harn_vm::llm::receipts::{TOOL_CALL_RECEIPT_EXECUTORS, TOOL_CALL_RECEIPT_STATUSES};
use super::constants::*;
use super::support::*;
use super::values::*;
pub(super) const PYTHON_INIT_STUB: &str = "from .harn_protocol import * # noqa: F401,F403\n";
pub(super) fn generate_python() -> String {
let mut out = String::new();
out.push_str("# GENERATED by `harn dump-protocol-artifacts` - do not edit by hand.\n");
out.push_str("# Source: Harn adapter schemas and Rust wire vocabulary.\n");
out.push_str("\"\"\"Python bindings for Harn's host/integrator protocol surface.\n\n");
out.push_str("Mirrors the TypeScript and Swift artifacts generated alongside this\n");
out.push_str("module. Field names match the wire JSON (camelCase) so dataclass\n");
out.push_str("instances round-trip through ``json.dumps(asdict(obj))`` without a\n");
out.push_str("custom encoder. Optional fields default to ``None`` and are stripped\n");
out.push_str("by :func:`to_wire`.\n");
out.push_str("\"\"\"\n\n");
out.push_str("from __future__ import annotations\n\n");
out.push_str("from dataclasses import asdict, dataclass, field, fields, is_dataclass\n");
out.push_str("from enum import Enum\n");
out.push_str("from typing import Any, Dict, List, Mapping, Optional, Type, TypeVar, Union\n\n");
out.push_str("__all__ = [\n");
let public_names = python_public_names();
for name in &public_names {
out.push_str(" ");
out.push_str(&json_string_literal(name));
out.push_str(",\n");
}
out.push_str("]\n\n");
for (name, value) in [
("HARN_PROTOCOL_ARTIFACT_VERSION", env!("CARGO_PKG_VERSION")),
("HARN_AGENT_EVENT_METHOD", HARN_AGENT_EVENT_METHOD),
("HARN_PROVIDER_CATALOG_METHOD", HARN_PROVIDER_CATALOG_METHOD),
("ACP_SCHEMA_COMPATIBILITY", ACP_SCHEMA_COMPATIBILITY),
("A2A_PROTOCOL_VERSION", A2A_PROTOCOL_VERSION),
("MCP_PROTOCOL_VERSION", MCP_PROTOCOL_VERSION),
("MCP_STABLE_PROTOCOL_VERSION", MCP_PROTOCOL_VERSION),
("MCP_DRAFT_PROTOCOL_VERSION", MCP_DRAFT_PROTOCOL_VERSION),
(
"MCP_FINAL_2026_PROTOCOL_VERSION",
MCP_FINAL_2026_PROTOCOL_VERSION,
),
(
"MCP_JSON_SCHEMA_2020_12_DIALECT",
MCP_JSON_SCHEMA_2020_12_DIALECT,
),
(
"MCP_INPUT_REQUIRED_RESULT_TYPE",
MCP_INPUT_REQUIRED_RESULT_TYPE,
),
(
"MCP_UNSUPPORTED_PROTOCOL_VERSION_ERROR_MESSAGE",
MCP_UNSUPPORTED_PROTOCOL_VERSION_ERROR_MESSAGE,
),
] {
out.push_str(&format!("{name}: str = {}\n", json_string_literal(value)));
}
out.push_str(&format!(
"MCP_UNSUPPORTED_PROTOCOL_VERSION_ERROR_CODE: int = {MCP_UNSUPPORTED_PROTOCOL_VERSION_ERROR_CODE}\n"
));
out.push('\n');
out.push_str(&py_const_tuple("ACP_AGENT_METHODS", ACP_AGENT_METHODS));
out.push_str(&py_const_tuple("ACP_CLIENT_METHODS", ACP_CLIENT_METHODS));
out.push_str(&py_const_tuple(
"ACP_AGENT_NOTIFICATIONS",
ACP_AGENT_NOTIFICATIONS,
));
let session_updates = all_acp_session_updates();
out.push_str(&py_const_tuple_owned(
"ACP_SESSION_UPDATES",
&session_updates,
));
out.push_str(&py_const_tuple(
"HARN_ACP_SESSION_UPDATE_EXTENSIONS",
HARN_SESSION_UPDATE_EXTENSIONS,
));
out.push_str(&py_const_tuple(
"HARN_AGENT_EVENT_KINDS",
HARN_AGENT_EVENT_KINDS,
));
out.push_str(&py_const_tuple(
"ACP_CONTENT_BLOCK_TYPES",
ACP_CONTENT_BLOCK_TYPES,
));
out.push_str(&py_const_tuple_owned("ACP_TOOL_KINDS", &tool_kind_values()));
out.push_str(&py_const_tuple_owned(
"ACP_TOOL_CALL_STATUSES",
&tool_call_status_values(),
));
out.push_str(&py_const_tuple_owned(
"HARN_TOOL_CALL_ERROR_CATEGORIES",
&tool_call_error_category_values(),
));
out.push_str(&py_const_tuple_owned(
"HARN_SIDE_EFFECT_LEVELS",
&side_effect_level_values(),
));
out.push_str(&py_const_tuple_owned(
"HARN_WORKER_STATUSES",
&worker_status_values(),
));
out.push_str(&py_const_tuple(
"HARN_TOOL_CALL_RECEIPT_STATUSES",
TOOL_CALL_RECEIPT_STATUSES,
));
out.push_str(&py_const_tuple(
"HARN_TOOL_CALL_RECEIPT_EXECUTORS",
TOOL_CALL_RECEIPT_EXECUTORS,
));
out.push_str(&py_const_tuple(
"HARN_TOOL_LIFECYCLE_EXTENSION_FIELDS",
HARN_TOOL_LIFECYCLE_EXTENSION_FIELDS,
));
out.push_str(&py_const_tuple(
"HARN_CONTENT_EXTENSION_FIELDS",
HARN_CONTENT_EXTENSION_FIELDS,
));
out.push_str(&py_const_tuple("A2A_METHODS", A2A_METHODS));
out.push_str(&py_const_tuple("A2A_TASK_STATES", A2A_TASK_STATES));
out.push_str(&py_const_tuple(
"A2A_TASK_EVENT_TYPES",
A2A_TASK_EVENT_TYPES,
));
out.push_str(&py_const_tuple(
"MCP_PROTOCOL_VERSIONS",
MCP_PROTOCOL_VERSIONS,
));
out.push_str(&py_const_tuple("MCP_METHODS", MCP_METHODS));
out.push_str(&py_const_tuple(
"MCP_REQUIRED_METADATA_KEYS",
MCP_REQUIRED_METADATA_KEYS,
));
out.push_str(&py_const_tuple("MCP_METADATA_KEYS", MCP_METADATA_KEYS));
out.push_str(&py_const_tuple(
"MCP_STANDARD_HTTP_HEADERS",
MCP_STANDARD_HTTP_HEADERS,
));
out.push_str(&py_const_tuple(
"MCP_CACHE_RESULT_FIELDS",
MCP_CACHE_RESULT_FIELDS,
));
out.push_str(&py_const_tuple("MCP_CACHE_SCOPES", MCP_CACHE_SCOPES));
out.push_str(&py_const_tuple("MCP_RESULT_TYPES", MCP_RESULT_TYPES));
out.push_str(&py_const_tuple("MCP_LOGGING_LEVELS", MCP_LOGGING_LEVELS));
out.push_str(&py_str_enum("ACPAgentMethod", ACP_AGENT_METHODS));
out.push_str(&py_str_enum("ACPClientMethod", ACP_CLIENT_METHODS));
out.push_str(&py_str_enum(
"ACPAgentNotification",
ACP_AGENT_NOTIFICATIONS,
));
out.push_str(&py_str_enum_owned("ACPSessionUpdate", &session_updates));
out.push_str(&py_str_enum_owned("ACPToolKind", &tool_kind_values()));
out.push_str(&py_str_enum_owned(
"ACPToolCallStatus",
&tool_call_status_values(),
));
out.push_str(&py_str_enum_owned(
"HarnToolCallErrorCategory",
&tool_call_error_category_values(),
));
out.push_str(&py_str_enum_owned(
"HarnSideEffectLevel",
&side_effect_level_values(),
));
out.push_str(&py_str_enum_owned(
"HarnWorkerStatus",
&worker_status_values(),
));
out.push_str(&py_str_enum(
"ToolCallReceiptStatus",
TOOL_CALL_RECEIPT_STATUSES,
));
out.push_str(&py_str_enum(
"ToolCallReceiptExecutor",
TOOL_CALL_RECEIPT_EXECUTORS,
));
out.push_str(&py_str_enum("A2ATaskState", A2A_TASK_STATES));
out.push_str(&py_str_enum("A2ATaskEventType", A2A_TASK_EVENT_TYPES));
out.push_str(&py_str_enum("MCPCacheScope", MCP_CACHE_SCOPES));
out.push_str(&py_str_enum("MCPResultType", MCP_RESULT_TYPES));
out.push_str(&py_str_enum("MCPLoggingLevel", MCP_LOGGING_LEVELS));
out.push_str(PYTHON_TYPE_DEFINITIONS);
out
}
pub(super) const PYTHON_TYPE_DEFINITIONS: &str = r#"
JsonValue = Union[None, bool, int, float, str, List[Any], Dict[str, Any]]
JsonObject = Dict[str, JsonValue]
JsonRpcId = Union[int, str, None]
_T = TypeVar("_T", bound="_HarnDataclass")
class _HarnDataclass:
"""Mixin providing wire-faithful ``to_wire`` / ``from_wire`` helpers.
``to_wire`` strips fields that are ``None`` so the JSON envelope stays
minimal and matches what the Rust adapters emit. ``from_wire`` accepts a
decoded JSON object plus any unknown extension fields, ignoring keys
that the static schema does not yet know about so additive enum or
struct extensions stay forward-compatible.
"""
@classmethod
def from_wire(cls: Type[_T], data: Mapping[str, Any]) -> _T:
if not isinstance(data, Mapping):
raise TypeError(f"{cls.__name__}.from_wire expected a mapping, got {type(data).__name__}")
accepted = {f.name for f in fields(cls)} # type: ignore[arg-type]
kwargs = {key: value for key, value in data.items() if key in accepted}
return cls(**kwargs)
def to_wire(self) -> JsonObject:
if not is_dataclass(self):
raise TypeError("to_wire requires a dataclass instance")
return _strip_none(asdict(self))
def _strip_none(value: Any) -> Any:
if isinstance(value, dict):
return {key: _strip_none(item) for key, item in value.items() if item is not None}
if isinstance(value, list):
return [_strip_none(item) for item in value]
if isinstance(value, Enum):
return value.value
return value
@dataclass
class ACPError(_HarnDataclass):
code: int
message: str
data: Optional[JsonValue] = None
@dataclass
class ACPRequest(_HarnDataclass):
id: JsonRpcId
method: str
params: Optional[JsonValue] = None
jsonrpc: str = "2.0"
@dataclass
class ACPResponse(_HarnDataclass):
id: JsonRpcId
result: Optional[JsonValue] = None
error: Optional[ACPError] = None
jsonrpc: str = "2.0"
@dataclass
class ACPNotification(_HarnDataclass):
method: str
params: Optional[JsonValue] = None
jsonrpc: str = "2.0"
ACPMessage = Union[ACPRequest, ACPResponse, ACPNotification]
def is_request(message: Mapping[str, Any]) -> bool:
return "id" in message and "method" in message
def is_response(message: Mapping[str, Any]) -> bool:
return "id" in message and "method" not in message
def is_notification(message: Mapping[str, Any]) -> bool:
return "id" not in message and "method" in message
@dataclass
class HarnExtensionMeta(_HarnDataclass):
harn: Optional[JsonObject] = None
@dataclass
class ACPContentBlock(_HarnDataclass):
type: str
text: Optional[str] = None
_meta: Optional[HarnExtensionMeta] = None
@dataclass
class HarnToolLifecycleMeta(_HarnDataclass):
audit: Optional[JsonValue] = None
durationMs: Optional[float] = None
error: Optional[str] = None
errorCategory: Optional[str] = None
executionDurationMs: Optional[float] = None
executor: Optional[JsonValue] = None
parsing: Optional[bool] = None
rawInputPartial: Optional[str] = None
@dataclass
class ToolCallReceipt(_HarnDataclass):
schema_version: int
session_id: str
tool_call_id: str
tool_name: str
iteration: int
emit_order: int
status: str
duration_ms: int
args_hash: str
audit: JsonValue
emitted_at: str
run_id: Optional[str] = None
turn_index: Optional[int] = None
reason: Optional[str] = None
kind: Optional[str] = None
executor: Optional[str] = None
error_category: Optional[str] = None
result_hash: Optional[str] = None
model: Optional[str] = None
provider: Optional[str] = None
def to_wire(self) -> JsonObject:
return asdict(self)
@dataclass
class ACPToolCall(_HarnDataclass):
sessionUpdate: str
toolCallId: str
title: str
kind: Optional[str] = None
status: Optional[str] = None
content: Optional[List[ACPContentBlock]] = None
locations: Optional[List[JsonValue]] = None
rawInput: Optional[JsonValue] = None
rawOutput: Optional[JsonValue] = None
_meta: Optional[HarnExtensionMeta] = None
@dataclass
class ACPToolCallUpdate(_HarnDataclass):
sessionUpdate: str
toolCallId: str
title: Optional[str] = None
kind: Optional[str] = None
status: Optional[str] = None
content: Optional[List[ACPContentBlock]] = None
locations: Optional[List[JsonValue]] = None
rawInput: Optional[JsonValue] = None
rawOutput: Optional[JsonValue] = None
_meta: Optional[HarnExtensionMeta] = None
@dataclass
class ACPSessionUpdateEnvelope(_HarnDataclass):
sessionUpdate: str
content: Optional[JsonValue] = None
messageId: Optional[str] = None
entries: Optional[List[JsonValue]] = None
keptTurnCount: Optional[int] = None
removedTurnCount: Optional[int] = None
newTipTurnId: Optional[str] = None
reason: Optional[str] = None
toolCallId: Optional[str] = None
title: Optional[str] = None
kind: Optional[str] = None
status: Optional[str] = None
rawInput: Optional[JsonValue] = None
rawOutput: Optional[JsonValue] = None
_meta: Optional[HarnExtensionMeta] = None
@dataclass
class ACPSessionUpdateParams(_HarnDataclass):
sessionId: str
update: ACPSessionUpdateEnvelope
@dataclass
class ACPSessionUpdateNotification(_HarnDataclass):
params: ACPSessionUpdateParams
method: str = "session/update"
jsonrpc: str = "2.0"
@dataclass
class HarnAgentEventNotification(_HarnDataclass):
params: JsonObject
method: str = HARN_AGENT_EVENT_METHOD
jsonrpc: str = "2.0"
@dataclass
class HarnToolArgSchema(_HarnDataclass):
path_params: List[str] = field(default_factory=list)
arg_aliases: Dict[str, str] = field(default_factory=dict)
required: List[str] = field(default_factory=list)
@dataclass
class HarnToolAnnotations(_HarnDataclass):
kind: str
side_effect_level: str
arg_schema: HarnToolArgSchema
capabilities: Dict[str, List[str]] = field(default_factory=dict)
emits_artifacts: bool = False
result_readers: List[str] = field(default_factory=list)
inline_result: bool = False
@dataclass
class A2AMessage(_HarnDataclass):
id: str
role: str
parts: List[JsonValue] = field(default_factory=list)
@dataclass
class A2ATaskStatus(_HarnDataclass):
state: str
message: Optional[A2AMessage] = None
timestamp: Optional[str] = None
@dataclass
class A2ATask(_HarnDataclass):
id: str
status: A2ATaskStatus
contextId: Optional[str] = None
history: Optional[List[A2AMessage]] = None
artifacts: Optional[List[JsonValue]] = None
metadata: Optional[JsonObject] = None
MCPJsonSchema202012 = JsonObject
MCPRequestMeta = JsonObject
MCPHTTPHeaders = Dict[str, str]
@dataclass
class MCPImplementation(_HarnDataclass):
name: str
version: str
title: Optional[str] = None
description: Optional[str] = None
websiteUrl: Optional[str] = None
@dataclass
class MCPCacheHints(_HarnDataclass):
ttlMs: int
cacheScope: str
@dataclass
class MCPDiscoverResult(_HarnDataclass):
resultType: str
supportedVersions: List[str]
capabilities: JsonObject
serverInfo: MCPImplementation
instructions: Optional[str] = None
_meta: Optional[JsonObject] = None
@dataclass
class MCPInputRequiredResult(_HarnDataclass):
resultType: str
inputRequests: Optional[JsonObject] = None
requestState: Optional[str] = None
_meta: Optional[JsonObject] = None
@dataclass
class MCPUnsupportedProtocolVersionErrorData(_HarnDataclass):
requested: str
supported: List[str]
@dataclass
class MCPUnsupportedProtocolVersionError(_HarnDataclass):
error: ACPError
id: JsonRpcId = None
jsonrpc: str = "2.0"
@dataclass
class MCPTool(_HarnDataclass):
name: str
inputSchema: MCPJsonSchema202012
title: Optional[str] = None
description: Optional[str] = None
outputSchema: Optional[MCPJsonSchema202012] = None
annotations: Optional[JsonObject] = None
@dataclass
class MCPResource(_HarnDataclass):
uri: str
name: str
title: Optional[str] = None
description: Optional[str] = None
mimeType: Optional[str] = None
@dataclass
class MCPResourceTemplate(_HarnDataclass):
uriTemplate: str
name: str
title: Optional[str] = None
description: Optional[str] = None
mimeType: Optional[str] = None
@dataclass
class MCPPrompt(_HarnDataclass):
name: str
title: Optional[str] = None
description: Optional[str] = None
arguments: Optional[List[JsonObject]] = None
"#;
pub(super) fn python_public_names() -> Vec<String> {
let names = [
"HARN_PROTOCOL_ARTIFACT_VERSION",
"HARN_AGENT_EVENT_METHOD",
"HARN_PROVIDER_CATALOG_METHOD",
"ACP_SCHEMA_COMPATIBILITY",
"A2A_PROTOCOL_VERSION",
"MCP_PROTOCOL_VERSION",
"MCP_STABLE_PROTOCOL_VERSION",
"MCP_DRAFT_PROTOCOL_VERSION",
"MCP_FINAL_2026_PROTOCOL_VERSION",
"MCP_JSON_SCHEMA_2020_12_DIALECT",
"MCP_INPUT_REQUIRED_RESULT_TYPE",
"MCP_UNSUPPORTED_PROTOCOL_VERSION_ERROR_CODE",
"MCP_UNSUPPORTED_PROTOCOL_VERSION_ERROR_MESSAGE",
"ACP_AGENT_METHODS",
"ACP_CLIENT_METHODS",
"ACP_AGENT_NOTIFICATIONS",
"ACP_SESSION_UPDATES",
"HARN_ACP_SESSION_UPDATE_EXTENSIONS",
"HARN_AGENT_EVENT_KINDS",
"ACP_CONTENT_BLOCK_TYPES",
"ACP_TOOL_KINDS",
"ACP_TOOL_CALL_STATUSES",
"HARN_TOOL_CALL_ERROR_CATEGORIES",
"HARN_SIDE_EFFECT_LEVELS",
"HARN_WORKER_STATUSES",
"HARN_TOOL_CALL_RECEIPT_STATUSES",
"HARN_TOOL_CALL_RECEIPT_EXECUTORS",
"HARN_TOOL_LIFECYCLE_EXTENSION_FIELDS",
"HARN_CONTENT_EXTENSION_FIELDS",
"A2A_METHODS",
"A2A_TASK_STATES",
"A2A_TASK_EVENT_TYPES",
"MCP_PROTOCOL_VERSIONS",
"MCP_METHODS",
"MCP_REQUIRED_METADATA_KEYS",
"MCP_METADATA_KEYS",
"MCP_STANDARD_HTTP_HEADERS",
"MCP_CACHE_RESULT_FIELDS",
"MCP_CACHE_SCOPES",
"MCP_RESULT_TYPES",
"MCP_LOGGING_LEVELS",
"ACPAgentMethod",
"ACPClientMethod",
"ACPAgentNotification",
"ACPSessionUpdate",
"ACPToolKind",
"ACPToolCallStatus",
"HarnToolCallErrorCategory",
"HarnSideEffectLevel",
"HarnWorkerStatus",
"ToolCallReceiptStatus",
"ToolCallReceiptExecutor",
"A2ATaskState",
"A2ATaskEventType",
"MCPCacheScope",
"MCPResultType",
"MCPLoggingLevel",
"ACPError",
"ACPRequest",
"ACPResponse",
"ACPNotification",
"ACPMessage",
"HarnExtensionMeta",
"ACPContentBlock",
"HarnToolLifecycleMeta",
"ToolCallReceipt",
"ACPToolCall",
"ACPToolCallUpdate",
"ACPSessionUpdateEnvelope",
"ACPSessionUpdateParams",
"ACPSessionUpdateNotification",
"HarnAgentEventNotification",
"HarnToolArgSchema",
"HarnToolAnnotations",
"A2AMessage",
"A2ATaskStatus",
"A2ATask",
"MCPJsonSchema202012",
"MCPRequestMeta",
"MCPHTTPHeaders",
"MCPImplementation",
"MCPCacheHints",
"MCPDiscoverResult",
"MCPInputRequiredResult",
"MCPUnsupportedProtocolVersionErrorData",
"MCPUnsupportedProtocolVersionError",
"MCPTool",
"MCPResource",
"MCPResourceTemplate",
"MCPPrompt",
"JsonValue",
"JsonObject",
"JsonRpcId",
"is_request",
"is_response",
"is_notification",
];
names.into_iter().map(String::from).collect()
}
pub(super) fn py_const_tuple(name: &str, values: &[&str]) -> String {
py_const_tuple_owned(name, &strs_to_strings(values))
}
pub(super) fn py_const_tuple_owned(name: &str, values: &[String]) -> String {
let mut out = format!("{name}: tuple = (\n");
for value in values {
out.push_str(" ");
out.push_str(&json_string_literal(value));
out.push_str(",\n");
}
out.push_str(")\n");
out
}
pub(super) fn py_str_enum(name: &str, values: &[&str]) -> String {
py_str_enum_owned(name, &strs_to_strings(values))
}
pub(super) fn py_str_enum_owned(name: &str, values: &[String]) -> String {
let mut out = format!("\n\nclass {name}(str, Enum):\n");
for value in values {
let case = py_enum_member_name(value);
out.push_str(" ");
out.push_str(&case);
out.push_str(" = ");
out.push_str(&json_string_literal(value));
out.push('\n');
}
out
}
pub(super) fn py_enum_member_name(value: &str) -> String {
let mut out = String::new();
for ch in value.chars() {
if ch.is_alphanumeric() {
out.extend(ch.to_uppercase());
} else {
out.push('_');
}
}
let trimmed = out.trim_matches('_').to_string();
let collapsed = collapse_repeated_underscores(&trimmed);
if collapsed.is_empty() {
"UNKNOWN".to_string()
} else if collapsed
.chars()
.next()
.is_some_and(|ch| ch.is_ascii_digit())
{
format!("_{collapsed}")
} else {
collapsed
}
}