ROAM Python SDK

Python bindings for the Object Agent Mapper (ROAM). Install via pip:

pip install roam-sdk

Quick Start

from roam_sdk import RoamClient

client = RoamClient(address="localhost:50051", api_key="your-key")
client.connect()
client.register("my-agent", "1.0.0", mode=1)  # 1 = CODE_FIRST

RoamClient

Main entry point for interacting with the ROAM gRPC backend.

Constructor

RoamClient(address: str = "localhost:50051", api_key: Optional[str] = None)

Parameter	Type	Default	Description
`address`	str	`"localhost:50051"`	gRPC backend address
`api_key`	Optional[str]	`None`	API key injected as `x-roam-api-key` metadata

connect()

Establishes the gRPC channel. Must be called before any other operation.

client.connect()  # returns True on success

register(agent_id, version, mode)

Registers this agent with the backend. Returns the gRPC ConnectResponse.

resp = client.register("my-agent", "1.0.0", mode=1)
print(resp.session_id)

Parameter	Type	Description
`agent_id`	str	Unique identifier for the agent
`version`	str	Semver string
`mode`	int	`0`=DATA_FIRST, `1`=CODE_FIRST, `2`=HYBRID

set_query_context(**kwargs)

Attaches per-request context propagated as gRPC metadata on every query.

client.set_query_context(
    organization_id="org-123",
    user_id="user-456",
    tool_name="search_orders",
    tool_intent="lookup recent orders",
    grants=["read:orders"],
    domain_tags=["commerce"],
    table_names=["orders"],
)

kwarg	gRPC header
`organization_id`	`x-roam-organization-id`
`user_id`	`x-roam-user-id`
`tool_name`	`x-roam-tool-name`
`tool_intent`	`x-roam-tool-intent`
`grants`	`x-roam-grants` (comma-joined)
`domain_tags`	`x-roam-domain-tags` (comma-joined)
`table_names`	`x-roam-table-names` (comma-joined)
`runtime_augmentation_id`	`x-roam-runtime-augmentation-id`
`runtime_augmentation_key`	`x-roam-runtime-augmentation-key`

clear_query_context()

Resets all query context fields.

register_model(model_class)

Registers a SQLAlchemy model class (must inherit RoamDeclarativeBase) with the agent. Raises ValueError in DATA_FIRST mode.

execute_query(sql)

Executes a SQL query string against the backend. Returns the gRPC response.

result = client.execute_query("SELECT id, name FROM users LIMIT 10")

RoamDeclarativeBase

SQLAlchemy mixin that enables ROAM schema introspection on your models.

from sqlalchemy.orm import DeclarativeBase
from roam_sdk import RoamDeclarativeBase

class Base(RoamDeclarativeBase, DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "users"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str]
    email: Mapped[str]

to_roam_schema() → dict

Introspects the SQLAlchemy model and returns a JSON Schema–compatible dict suitable for LLM tool definitions.

schema = User.to_roam_schema()
# { "name": "users", "parameters": { "type": "object", "properties": { ... } } }

from_agent_tool_call(**kwargs) → instance

Factory method to instantiate a model from agent-supplied kwargs with validation.

@wrap_tool_call decorator

Wraps a Python function as an OAM-compatible tool with injected metadata.

from roam_sdk.decorators import wrap_tool_call
from pydantic import BaseModel

class SearchInput(BaseModel):
    query: str
    limit: int = 10

@wrap_tool_call(name="search_orders", description="Search orders by keyword", schema=SearchInput)
def search_orders(query: str, limit: int = 10):
    ...  # implementation

# Access metadata
print(search_orders.oam_tool_def)
# { "name": "search_orders", "description": "...", "input_schema": { ... } }

gRPC stubs: Python bindings require generated proto files. Run roam proto gen or regenerate manually:

python -m grpc_tools.protoc -I../../roam-proto/src \
  --python_out=. --grpc_python_out=. \
  ../../roam-proto/src/v1/agent/service.proto