Robonix Python 公共接口

本页只列出软件包开发者直接使用的公共入口。签名、默认值和源码链接由 Sphinx 从当前 Robonix 源码生成；概念与完整示例见《Robonix 开发者指南》。

能力提供方

Primitive、Service 和 Skill 共享下面的方法。这里以公开的 Primitive 类展示；三者的区别是向 Atlas 注册的提供方类型，以及技能的 延迟激活行为。

class robonix_api.Primitive(id: str, namespace: str, *, pkg_root: Path | None = None, md_path: str | None = None)

Bases: _ProviderBase

A hardware or data-source capability provider.

Examples include a camera, lidar, or CAN chassis driver.

Example:

primitive_cam = Primitive(
    id="webots_tiago_camera_front",
    namespace="robonix/primitive/camera",
)

attach_grpc_servicer(contract_id: str, servicer, *, description: str = '') → None

Attach an already-built Servicer instance for contract_id. Use this for multi-method services; for single-method handlers prefer @provider.provides_grpc(…).

bootstrap() → None

Non-blocking setup. Idempotent on re-entry.

connect_capability(provider: CapabilityProvider | Capability, contract_id: str, transport: Transport | str | int) → Channel

Open a channel to another CapabilityProvider’s Capability. provider may be a CapabilityProvider (from ATLAS.query_*) or a Capability (from ATLAS.find_capability); both carry the provider id.

create_publisher(contract_id: str, *, topic: str, msg_type: type | str, qos: str | int = 'best_effort', declare: bool = True, description: str = '')

create_subscription(contract_id: str, *, topic: str, msg_type: type | str, callback: Callable[[Any], None], qos: str | int = 'best_effort', declare: bool = True)

create_subscription_from_channel(channel: Channel, *, msg_type: type | str, callback: Callable[[Any], None])

declare_capability(contract_id: str, endpoint: str, transport: Transport | str | int, params: GrpcParams | Ros2Params | McpParams | None = None, description: str = '') → str

Declare a Capability for contract_id on this CapabilityProvider. description is the instance-specific natural-language string Pilot/LLM sees; empty means “use the contract’s generic description from the TOML at consume time” (the two are merged, not picked-one-of). Namespace alignment is advisory: a regular contract outside this provider’s primary namespace emits a warning but is still declared; contracts marked cross_namespace do not.

declare_grpc(contract_id: str, endpoint: str, service_name: str, method: str, proto_file: str = 'robonix_contracts.proto', description: str = '') → str

declare_mcp(contract_id: str, endpoint: str, input_schema_json: str = '{}', description: str = '') → str

declare_ros2_service(contract_id: str, service: str, *, description: str = '') → str

Declare a ROS 2 service endpoint for an rpc contract over ROS 2.

declare_ros2_topic(contract_id: str, topic: str, *, qos: str = 'best_effort', description: str = '') → str

Declare a ROS 2 topic endpoint for a topic_in / topic_out contract.

emit(contract_id: str, msg: Any) → None

grpc(contract_id: str, *, description: str = '')

Bind a handler to contract_id’s generated gRPC Servicer. Description is taken from the function’s docstring unless description= is passed explicitly.

mcp(contract_id: str, *, description: str = '')

Register an MCP tool bound to contract_id. The MCP-server- side tool name is the contract_id’s leaf segment — same value executor’s dispatch derives — so there is no overridable name=. The natural-language description is taken from the wrapped function’s docstring unless description= is passed explicitly.

property mcp_endpoint: str

on_activate(fn: Callable[[], Any]) → Callable[[], Any]

INACTIVE -> ACTIVE. Acquire hot runtime resources (threads, models, ROS subs, hardware fds). Optional for Primitives / Services (framework auto-promotes); REQUIRED for Skills.

on_deactivate(fn: Callable[[], Any]) → Callable[[], Any]

ACTIVE -> INACTIVE. Release hot resources, keep config / atlas registration.

on_init(fn: Callable[[dict], Any]) → Callable[[dict], Any]

REGISTERED -> INACTIVE. Parse config, validate dependencies, bind logical device. NO hot runtime resources yet.

on_shutdown(fn: Callable[[], Any]) → Callable[[], Any]

any -> TERMINATED. Last-chance cleanup before process exit.

provides_grpc(contract_id: str, *, description: str = '')

Bind a handler to contract_id’s generated gRPC Servicer. Description is taken from the function’s docstring unless description= is passed explicitly.

provides_mcp(contract_id: str, *, description: str = '')

Register an MCP tool bound to contract_id. The MCP-server- side tool name is the contract_id’s leaf segment — same value executor’s dispatch derives — so there is no overridable name=. The natural-language description is taken from the wrapped function’s docstring unless description= is passed explicitly.

resolve_host_ip(target_ip: str) → str | None

ip route get <target> -> src field. Used by drivers (e.g. mid360) that need to bake the host’s IP into a vendor config.

run() → None

Blocking. Calls bootstrap() then signal.pause()-equivalents until SIGTERM / SIGINT.

spawn(argv, *, env: dict | None = None, log: str | Path | None = None, cwd: Path | None = None)

property state: LifecycleState

use_mcp_app(app) → None

wait_for_topic(topic: str, msg_type: str | type, timeout_s: float = 30.0) → bool

构造函数与类说明见 robonix_api.Primitive、 robonix_api.Service 和 robonix_api.Skill。

能力目录

robonix_api.ATLAS 是公开的能力目录单例。下面列出开发者用于发现能力、 查询能力约定和建立通道的方法。

class robonix_api.atlas._Atlas(endpoint: str | None = None)

Bases: object

Singleton facade for the atlas gRPC client.

Exposed as the module-level ATLAS constant. Lazy-connects on first use, reads $ROBONIX_ATLAS (default 127.0.0.1:50051).

register_primitive(id: str, namespace: str, capability_md_path: str = '') → str

register_service(id: str, namespace: str, capability_md_path: str = '') → str

register_skill(id: str, namespace: str, capability_md_path: str = '') → str

unregister(id: str) → bool

heartbeat(id: str) → None

start_heartbeat(id: str, period_s: float = 30.0, stop: Event | None = None) → Thread

Background daemon thread that pings Heartbeat every period_s seconds. Returns the thread for caller bookkeeping (or to ignore).

If stop is given, the loop polls it via stop.wait(period_s) and exits cleanly when set — providers should pass their teardown Event so the heartbeat thread stops trying to ping atlas after _set_state(TERMINATED). Default None keeps the legacy unbounded loop (interpreter-exit relies on daemon=True).

declare_capability(provider_id: str, contract_id: str, transport: Transport | str | int, endpoint: str, params: GrpcParams | Ros2Params | McpParams | None = None, description: str = '') → str

Declare one Capability on a registered CapabilityProvider. Returns the authoritative endpoint Atlas stored (may differ from endpoint when Atlas rewrote on collision).

query(*, kind: Kind | str | int = Kind.UNSPECIFIED, id: str = '', contract_id: str = '', namespace_prefix: str = '', transport: Transport | str | int = Transport.UNSPECIFIED) → list[CapabilityProvider]

Generic Query. Kind=UNSPECIFIED returns all kinds; each Record’s kind field carries the actual kind so callers can demultiplex.

query_primitives(*, id: str = '', contract_id: str = '', namespace_prefix: str = '', transport: Transport | str | int = Transport.UNSPECIFIED) → list[CapabilityProvider]

query_services(*, id: str = '', contract_id: str = '', namespace_prefix: str = '', transport: Transport | str | int = Transport.UNSPECIFIED) → list[CapabilityProvider]

query_skills(*, id: str = '', contract_id: str = '', namespace_prefix: str = '', transport: Transport | str | int = Transport.UNSPECIFIED) → list[CapabilityProvider]

find_capability(*, contract_id: str = '', transport: Transport | str | int = Transport.UNSPECIFIED, provider_kind: Kind | str | int = Kind.UNSPECIFIED, provider_id: str = '', namespace_prefix: str = '') → list[Capability]

Flat consumer-facing list of Capabilities matching the filters. Walks Query() and flattens each provider’s matching capabilities[]. Atlas returns whole provider records when any one of their capabilities matches, so the client must apply the capability filters again while flattening. Each returned Capability already carries provider_id / provider_kind.

find_unique_capability(*, contract_id: str, transport: Transport | str | int = Transport.UNSPECIFIED, provider_kind: Kind | str | int = Kind.UNSPECIFIED, provider_id: str = '', namespace_prefix: str = '') → Capability

Like find_capability but expects exactly one match. Raises ValueError on 0 or >1 matches — for “I depend on THE camera/depth capability” wiring where ambiguity is a config bug.

query_contract(contract_id: str) → ContractDescriptor | None

list_contracts(namespace_prefix: str = '') → list[ContractDescriptor]

connect_capability(*, consumer_id: str, provider_id: str, contract_id: str, transport: Transport | str | int) → Channel

Open a consumer->provider edge. Returns a Channel context manager — with ATLAS.connect_capability(…) as ch: … auto-disconnects.

disconnect_capability(channel_id: str) → bool

inspect() → dict

数据类型

返回值和枚举见 robonix_api.atlas_types，包括 Capability、 CapabilityProvider、 Channel、 Transport 和 LifecycleState。