Python API reference

Generated from the source. The modules most people need are ascribe_link.models (result types), ascribe_link.processing (the function registry), and ascribe_link.app (create_app).

Application

Litestar application factory.

ascribe_link.app.create_app(specimens_dir: str | Path | None = None, mesh_functions: dict[str, collections.abc.Callable] | None = None, relay_mode: bool = False, enable_agent: bool = False, agent_model: str = 'claude-opus-4-8', agent_timeout: float = 300000.0) → Litestar[source]

Create and configure the Litestar application.

Parameters:
specimens_dir: str | Path | None = None

Path to the directory containing curated specimen bundles. If None, defaults to <repo_root>/specimens (relative to this module).

mesh_functions: dict[str, collections.abc.Callable] | None = None

Optional dict of {name: callable} processing functions to register.

relay_mode: bool = False

If True, enable federation hub for accepting worker connections. Workers can connect via WebSocket to register their specimens.

enable_agent: bool = False

If True, register the AI agent-based mesh generation function. Requires claude-agent-sdk to be installed.

agent_model: str = 'claude-opus-4-8'

Claude model to use for agent-based generation.

agent_timeout: float = 300000.0

Timeout in seconds for agent-based generation.

Results and metadata

Data models for Ascribe-Link.

class ascribe_link.models.FunctionInfo(name: str, schema: dict[str, Any] | None = None, return_type: str | None = None)[source]

Bases: object

Info about a registered processing function.

class ascribe_link.models.ImageResult(type: 'image' = 'image', width: int = 0, height: int = 0, channels: int = 1, dtype: str = 'uint8', data: str = '')[source]

Bases: object

2D image result.

type

Discriminator field.

Type:

Literal[“image”]

width

Image width in pixels.

Type:

int

height

Image height in pixels.

Type:

int

channels

Number of channels (1=grayscale, 3=RGB, 4=RGBA).

Type:

int

dtype

NumPy dtype string (e.g., “uint8”, “float32”).

Type:

str

data

Base64-encoded raw bytes (row-major, channels last).

Type:

str

classmethod from_numpy(arr: ndarray) → ImageResult[source]

Create from a NumPy array (H, W) or (H, W, C).

to_numpy() → ndarray[source]

Decode to NumPy array.

class ascribe_link.models.MeshResult(type: ~typing.Literal['mesh'] = 'mesh', vertices: list[float] = <factory>, indices: list[int] = <factory>, normals: list[float] | None = None)[source]

Bases: object

Mesh data result.

type

Discriminator field.

Type:

Literal[“mesh”]

vertices

Flat list of vertex coordinates [x1,y1,z1, x2,y2,z2, …].

Type:

list[float]

indices

Triangle indices (every 3 = one face).

Type:

list[int]

normals

Flat list of vertex normals [nx1,ny1,nz1, …].

Type:

list[float], optional

classmethod from_pyvista(mesh) → MeshResult[source]

Create from a PyVista mesh.

class ascribe_link.models.PointCloudResult(type: ~typing.Literal['point_cloud'] = 'point_cloud', points: list[float] = <factory>, colors: list[float] | None = None, scalars: list[float] | None = None, scalar_name: str | None = None)[source]

Bases: object

Point cloud data result.

type

Discriminator field.

Type:

Literal[“point_cloud”]

points

Flat list of point coordinates [x1,y1,z1, x2,y2,z2, …].

Type:

list[float]

colors

Flat list of RGB colors [r1,g1,b1, …] in range [0, 1].

Type:

list[float], optional

scalars

Per-point scalar values.

Type:

list[float], optional

scalar_name

Name of the scalar field.

Type:

str, optional

classmethod from_numpy(points: ndarray, colors: ndarray | None = None, scalars: ndarray | None = None, scalar_name: str | None = None) → PointCloudResult[source]

Create from NumPy arrays.

class ascribe_link.models.ProcessingRequest(function_name: str, args: list[~typing.Any] = <factory>, kwargs: dict[str, ~typing.Any] = <factory>, room_id: str = 'ascribe')[source]

Bases: object

Request to invoke a registered processing function.

class ascribe_link.models.ResultType(value)[source]

Bases: str, Enum

Type discriminator for processing results.

class ascribe_link.models.SpecimenListItem(id: str, display_name: str, description: str, type: ~ascribe_link.models.SpecimenType, thumbnail_url: str, tags: list[str] = <factory>, is_dynamic: bool = False)[source]

Bases: object

Lightweight specimen info for catalog listings.

class ascribe_link.models.SpecimenMetadata(id: str, display_name: str, description: str = '', type: ~ascribe_link.models.SpecimenType = SpecimenType.MESH, data_file: str = '', thumbnail_file: str = '', story_text: list[str] = <factory>, tags: list[str] = <factory>, schema: dict[str, ~typing.Any] | None = None, function_name: str | None = None)[source]

Bases: object

Metadata for a curated specimen.

class ascribe_link.models.SpecimenType(value)[source]

Bases: str, Enum

class ascribe_link.models.VolumeResult(type: ~typing.Literal['volume'] = 'volume', shape: list[int] = <factory>, dtype: str = 'float32', data: str = '', spacing: list[float] | None = None, origin: list[float] | None = None)[source]

Bases: object

Volumetric data result.

type

Discriminator field.

Type:

Literal[“volume”]

shape

Shape of the volume [depth, height, width] or [z, y, x].

Type:

list[int]

dtype

NumPy dtype string (e.g., “float32”, “uint8”).

Type:

str

data

Base64-encoded raw bytes of the volume array.

Type:

str

spacing

Voxel spacing [sz, sy, sx]. Defaults to [1, 1, 1].

Type:

list[float], optional

origin

Origin point [oz, oy, ox]. Defaults to [0, 0, 0].

Type:

list[float], optional

classmethod from_numpy(arr: ndarray, spacing: list[float] | None = None, origin: list[float] | None = None) → VolumeResult[source]

Create from a NumPy array.

to_numpy() → ndarray[source]

Decode to NumPy array.

ascribe_link.models.result_from_dict(data: dict[str, Any]) → MeshResult | VolumeResult | PointCloudResult | ImageResult[source]

Parse a result dictionary based on its ‘type’ field.

ascribe_link.models.result_to_dict(result: MeshResult | VolumeResult | PointCloudResult | ImageResult) → dict[str, Any][source]

Convert any result type to a dictionary.

Mesh utilities for agent-generated data.

These utilities are designed to be used by the AI agent when generating meshes with PyVista. They handle the common conversions needed to submit mesh data via the submit_mesh tool.

ascribe_link.mesh_utils.extract_mesh_data(mesh: pv.PolyData, include_normals: bool = True) → tuple[list[list[float]], list[int], list[float] | None][source]

Extract vertices, indices, and normals from a PyVista mesh.

Automatically triangulates the mesh if needed (converts quads to triangles) and converts numpy arrays to plain Python lists.

Parameters:
mesh : pv.PolyData

A PyVista mesh object (sphere, box, cylinder, or any PolyData).

include_normals : bool

If True, compute and return normals.

Returns:

  • vertices (list[float]) – Flat list of vertex coordinates [x, y, z, x, y, z, …].

  • indices (list[int]) – Flat list of vertex indices (every 3 = one triangle).

  • normals (list[float] | None) – Flat list of normal components [nx, ny, nz, nx, ny, nz, …] or None.

Example

>>> import pyvista as pv
>>> from ascribe_link.mesh_utils import extract_mesh_data
>>>
>>> sphere = pv.Sphere(radius=1.0)
>>> vertices, indices, normals = extract_mesh_data(sphere)
ascribe_link.mesh_utils.flatten_normals(normals_nested: list[list[float]]) → list[float][source]

Flatten nested normals [[nx, ny, nz], …] to [nx, ny, nz, nx, ny, nz, …].

Use this when you have normals from marching_cubes or other sources that return per-vertex normal vectors as nested lists.

Processing

Processing function registry.

Replaces the old MQTT-based function_map with a proper registry that supports schema generation and invocation via HTTP.

Functions can return various data types: - Mesh: (vertices, indices) tuple or MeshResult - Volume: VolumeResult or (array, spacing, origin) tuple - PointCloud: PointCloudResult or (points,) tuple - Image: ImageResult or 2D/3D numpy array

The return type is detected automatically or can be hinted with type annotations.

class ascribe_link.processing.FunctionRegistry[source]

Bases: object

Registry of processing functions.

get_specimen(function_name: str) → SpecimenMetadata | None[source]

Get specimen metadata for a registered function.

get_specimen_thumbnail(function_name: str) → str | None[source]

Get the thumbnail data URI for a registered specimen.

invoke(name: str, args: list | None = None, kwargs: dict | None = None) → MeshResult | VolumeResult | PointCloudResult | ImageResult[source]

Invoke a function (sync only) and return typed result.

For async functions, use invoke_async instead.

async invoke_async(name: str, args: list | None = None, kwargs: dict | None = None, reporter: ProgressReporter | None = None) → MeshResult | VolumeResult | PointCloudResult | ImageResult[source]

Invoke a function (async-aware) and return typed result.

list_specimens() → list[SpecimenMetadata][source]

List all registered specimens as SpecimenMetadata objects.

register(name: str | None = None, return_type: str | None = None, *, display_name: str | None = None, description: str | None = None, specimen_type: str | SpecimenType | None = None, tags: list[str] | None = None, thumbnail: str | None = None, story_text: list[str] | None = None) → Callable[source]

Decorator to register a processing function.

Parameters:
name : str, optional

Function name (defaults to function.__name__)

return_type : str, optional

Hint for return type: “mesh”, “volume”, “point_cloud”, “image”

display_name : str, optional

Human-readable name for specimen catalog. If provided, registers the function as a specimen.

description : str, optional

Description for specimen catalog.

specimen_type : str or SpecimenType, optional

Type of specimen: “mesh” or “volume”. Defaults to return_type or “mesh”.

tags : list[str], optional

Tags for categorization.

thumbnail : str, optional

Base64 data URI for thumbnail (e.g., “data:image/png;base64,…”).

story_text : list[str], optional

Story/narration text for the specimen.

register_function(func: Callable, name: str | None = None, return_type: str | None = None) → None[source]

Imperatively register a processing function.

register_specimen(func: Callable, display_name: str, *, name: str | None = None, description: str | None = None, return_type: str | None = None, specimen_type: str | SpecimenType | None = None, tags: list[str] | None = None, thumbnail: str | None = None, story_text: list[str] | None = None) → None[source]

Imperatively register a function as a specimen.

This is the non-decorator version for registering specimens.

class ascribe_link.processing.Range(min: float | int, max: float | int)[source]

Bases: object

Annotation for numeric parameters with min/max bounds.

Use with Annotated to generate JSON Schema with minimum/maximum:

from typing import Annotated from ascribe_link.processing import Range

def my_function(

segments: Annotated[int, Range(3, 128)] = 32

):

…

This will generate JSON Schema with:

{“type”: “number”, “minimum”: 3, “maximum”: 128, “default”: 32}

class ascribe_link.processing.RegisteredSpecimen(function_name: str, display_name: str, description: str = '', specimen_type: ~ascribe_link.models.SpecimenType = SpecimenType.MESH, tags: list[str] = <factory>, thumbnail: str | None = None, story_text: list[str] = <factory>)[source]

Bases: object

Metadata for a specimen registered via code.

ascribe_link.processing.create_schema(func: Callable) → dict[str, Any][source]

Generate a JSON Schema for a function’s parameters.

Parametric specimen generation functions.

These functions generate 3D data from parameters, suitable for dynamic specimens.

ascribe_link.parametric.generate_gaussian_volume(resolution: int = 64, sigma: float = 0.3) → VolumeResult[source]

Generate a 3D Gaussian blob centered in a unit cube.

Parameters:
resolution : int

Number of voxels per axis (clamped to [32, 256]).

sigma : float

Standard deviation of the Gaussian relative to the cube edge (clamped to [0.05, 1.0]).

Returns:

float32 volume, shape [resolution]*3, normalized so peak == 1.0.

Return type:

VolumeResult

ascribe_link.parametric.generate_sphere(radius: float = 1.0, resolution: int = 32) → MeshResult[source]

Generate a parametric sphere mesh.

Parameters:
radius : float

Sphere radius (0.5 to 5.0)

resolution : int

Number of divisions (8 to 128)

Returns:

Sphere mesh with vertices, indices, and normals

Return type:

MeshResult

ascribe_link.parametric.generate_torus(major_radius: float = 1.0, minor_radius: float = 0.3, segments: int = 32) → MeshResult[source]

Generate a parametric torus mesh.

Parameters:
major_radius : float

Distance from center to tube center (0.5 to 3.0)

minor_radius : float

Tube radius (0.1 to 1.0)

segments : int

Number of divisions (8 to 128)

Returns:

Torus mesh with vertices, indices, and normals

Return type:

MeshResult

Progress reporting primitives for dynamic specimen functions.

ProgressReporter is the interface that specimen functions depend on. The no-op default lets functions be called directly (in tests, REPL) without a job context. JobReporter (defined below) binds to a Job and appends messages to the job’s bounded message deque.

class ascribe_link.progress.JobReporter(job: Job)[source]

Bases: ProgressReporter

Reporter that appends messages to a specific Job’s deque.

report(text: str) → None[source]

Append a progress message. No-op in the base class.

class ascribe_link.progress.ProgressMessage(seq: int, text: str, ts: float)[source]

Bases: object

A single progress message in a job’s history.

class ascribe_link.progress.ProgressReporter[source]

Bases: object

Base reporter — no-op by default.

Functions declare a parameter of this type; FunctionRegistry.invoke_async will inject a real JobReporter if invoked under a Job, else this no-op.

report(text: str) → None[source]

Append a progress message. No-op in the base class.

Specimens and caching

Specimen store — manages a directory of curated specimens.

Each specimen is a subdirectory containing:

specimen.json — metadata (SpecimenMetadata fields) thumbnail.* — thumbnail image (png/jpg) <data_file> — mesh or volume data file

class ascribe_link.specimen_store.SpecimenStore(root: Path)[source]

Bases: object

Read-only store backed by a directory of specimen bundles.

reload() → None[source]

(Re-)scan the specimen directory and rebuild the cache.

ascribe_link.specimen_store.load_static_volume(spec_dir: Path, data_file: str) → VolumeResult[source]

Load a static volume specimen from disk (currently .npy + optional .json sidecar).

Parameters:
spec_dir : Path

Directory containing the specimen bundle.

data_file : str

Filename (within spec_dir) of the volume data, e.g. “data.npy”.

Return type:

VolumeResult

Result cache for multiplayer synchronization.

Caches processing results per room to ensure all peers receive identical data without redundant computation.

class ascribe_link.cache.CachedResult(room_id: str, function_name: str, params_hash: str, result: Any, timestamp: float, access_count: int = 0)[source]

Bases: object

A cached processing result.

class ascribe_link.cache.RoomResultCache(ttl_seconds: float = 300.0)[source]

Bases: object

Cache for processing results, keyed by room ID.

Each room can have one cached result at a time. When a new request comes in for a room, the old cached result for that room is invalidated (since the new specimen will replace the old one for all peers).

Thread-safe for concurrent access.

cleanup_expired() → int[source]

Remove expired entries from the cache.

Returns:

Number of entries removed

Return type:

int

clear() → None[source]

Clear all cached results.

get(room_id: str, function_name: str, params: dict[str, Any]) → Any | None[source]

Get a cached result if available.

Parameters:
room_id : str

Room identifier (e.g., “ascribe”)

function_name : str

Name of the processing function

params : dict

Function parameters (must match exactly)

Returns:

Cached result if found and valid, otherwise None. Stored objects may be raw Result instances or dicts; callers must handle either shape.

Return type:

Any | None

invalidate_room(room_id: str) → bool[source]

Explicitly invalidate the cached result for a room.

Parameters:
room_id : str

Room identifier

Returns:

True if an entry was removed, False if no entry existed

Return type:

bool

put(room_id: str, function_name: str, params: dict[str, Any], result: Any) → None[source]

Store a processing result in the cache.

Any previous cached result for this room is replaced.

Parameters:
room_id : str

Room identifier

function_name : str

Name of the processing function

params : dict

Function parameters

result : Any

Processing result to cache. May be a raw Result (MeshResult/VolumeResult/…) or a JSON-ready dict; consumers are responsible for normalizing on read.

stats() → dict[str, Any][source]

Get cache statistics.

Returns:

Statistics including entry count, room IDs, access counts

Return type:

dict

Job registry for dynamic specimen progress tracking.

Jobs are ephemeral — stored in-memory, not persisted. Each job tracks status, a bounded deque of progress messages, and the underlying asyncio.Task running the specimen function.

class ascribe_link.job_registry.Job(id: str, specimen_id: str, params: dict[str, ~typing.Any], room_id: str, status: ~typing.Literal['running', 'done', 'error'] = 'running', messages: ~collections.deque[~ascribe_link.progress.ProgressMessage] = <factory>, next_seq: int = 0, result: ~typing.Any = None, error: str | None = None, task: ~_asyncio.Task | None = None, created_at: float = <factory>, finished_at: float | None = None, federated_to: tuple[str, str] | None = None)[source]

Bases: object

A single specimen-load job.

Fields are mutated from the job’s own asyncio.Task (single-appender for messages) and read from poll handlers. collections.deque is safe for single-appender / multi-reader without explicit locking; poll handlers snapshot (next_seq, list(messages)) before serializing.

append_message(text: str) → ProgressMessage[source]

Append a new progress message and bump next_seq.

messages_since(since: int) → list[ProgressMessage][source]

Return only messages with seq > since (kept in the deque).

class ascribe_link.job_registry.JobRegistry(ttl_seconds: float = 300.0)[source]

Bases: object

In-memory job store with TTL sweeping for completed jobs.

async create(specimen_id: str, params: dict[str, Any], room_id: str) → Job[source]

Create a new running job and register it atomically.

async run_sweeper(interval: float = 30.0) → None[source]

Background task — sweep expired jobs every interval seconds.

async sweep_expired() → int[source]

Remove jobs whose finished_at is older than ttl_seconds.

Running jobs (status == “running”) are never swept even if old. Returns the number of jobs removed.

Wire format

Binary envelope wire format for MeshResult / VolumeResult.

Layout:

<4-byte little-endian uint32: preamble_length> <preamble_length bytes: UTF-8 JSON preamble> <raw bytes: one or more contiguous data blocks>

ascribe_link.envelope.decode_envelope(data: bytes) → MeshResult | VolumeResult[source]

Parse the binary envelope format back into a typed result.

ascribe_link.envelope.encode_envelope(result: MeshResult | VolumeResult) → bytes[source]

Serialize a result to the binary envelope format.

Federation

Federation layer for Ascribe-Link relay mode.

Allows multiple Ascribe-Link instances to federate through a relay: - Relay mode: Accepts WebSocket connections from workers, aggregates specimens - Worker mode: Connects outbound to a relay, registers local specimens

Architecture:

NERSC (worker) ──WS──▶ Neutral Host (relay) ◀──HTTP── Quest (client)

class ascribe_link.federation.FederationClient(relay_url: str, worker_id: str, on_request: Any = None)[source]

Bases: object

Worker-side client that connects to a relay.

Used when running in worker mode to register with a relay.

async start() → None[source]

Start the connection to the relay.

async stop() → None[source]

Stop the connection.

async update_functions(functions: list[dict[str, Any]]) → None[source]

Send updated function list to relay.

async update_specimens(specimens: list[dict[str, Any]]) → None[source]

Send updated specimen list to relay.

class ascribe_link.federation.FederationHub[source]

Bases: object

Relay-side hub that manages connected workers.

Used when running in relay mode to aggregate specimens from workers.

find_specimen_worker(specimen_id: str) → str | None[source]

Find which worker owns a specimen ID.

get_all_functions() → list[tuple[str, dict[str, Any]]][source]

Get all functions from all workers.

Returns list of (worker_id, function_dict) tuples.

get_all_specimens() → list[tuple[str, dict[str, Any]]][source]

Get all specimens from all workers.

Returns list of (worker_id, specimen_dict) tuples.

get_worker(worker_id: str) → WorkerInfo | None[source]

Get a worker by ID.

async proxy_request(worker_id: str, request_type: str, payload: dict[str, Any], timeout: float = 30.0) → dict[str, Any][source]

Send a request to a worker and wait for response.

Used to proxy data/thumbnail requests through to workers.

async register_worker(worker_id: str, websocket: WebSocket) → None[source]

Register a new worker connection.

store_response(worker_id: str, request_id: str, response: dict[str, Any]) → None[source]

Store a response from a worker (called by WS handler).

async unregister_worker(worker_id: str) → None[source]

Remove a worker connection.

async update_worker_functions(worker_id: str, functions: list[dict[str, Any]]) → None[source]

Update the function list for a worker.

async update_worker_specimens(worker_id: str, specimens: list[dict[str, Any]]) → None[source]

Update the specimen list for a worker.

class ascribe_link.federation.WorkerInfo(worker_id: str, websocket: WebSocket, specimens: list[dict[str, Any]] = <factory>, functions: list[dict[str, Any]] = <factory>)[source]

Bases: object

Information about a connected worker.

Agent generation

Firejail-based sandboxing for code execution.

Provides secure execution of untrusted Python code with: - Filesystem isolation (private tmpfs) - Network isolation (no network access) - Resource limits (memory, CPU time, wall time) - Capability dropping - Seccomp filtering

class ascribe_link.sandbox.SandboxConfig(memory_limit_mb: int = 4096, cpu_time_limit_secs: int = 300, wall_time_limit_secs: int = 600, network: bool = False, allow_paths: list[str] | None = None)[source]

Bases: object

Configuration for sandbox execution.

class ascribe_link.sandbox.SandboxResult(success: bool, stdout: str, stderr: str, return_code: int, output: Any | None = None, error: str | None = None)[source]

Bases: object

Result of sandboxed execution.

ascribe_link.sandbox.build_firejail_command(command: list[str], working_dir: Path, config: SandboxConfig, input_files: list[Path] | None = None) → list[str][source]

Build a Firejail command with appropriate sandboxing.

Parameters:
command : list[str]

The command to run (e.g., [“python3”, “script.py”])

working_dir : Path

Working directory (will be mounted read-write)

config : SandboxConfig

Sandbox configuration

input_files : list[Path], optional

Additional files to mount read-only

Returns:

Full Firejail command

Return type:

list[str]

ascribe_link.sandbox.get_firejail_version() → str | None[source]

Get Firejail version string.

ascribe_link.sandbox.is_firejail_available() → bool[source]

Check if Firejail is installed and accessible.

async ascribe_link.sandbox.run_sandboxed(code: str, working_dir: Path | None = None, config: SandboxConfig | None = None, input_files: list[Path] | None = None, python_executable: str = 'python3') → SandboxResult[source]

Run Python code in a Firejail sandbox.

Parameters:
code : str

Python code to execute

working_dir : Path, optional

Working directory. If None, creates a temp directory.

config : SandboxConfig, optional

Sandbox configuration. Uses defaults if None.

input_files : list[Path], optional

Additional files to make available (read-only)

python_executable : str

Python interpreter to use

Returns:

Execution result

Return type:

SandboxResult

async ascribe_link.sandbox.run_sandboxed_mesh_generation(code: str, config: SandboxConfig | None = None) → tuple[list[list[float]], list[int]][source]

Run mesh generation code in sandbox and extract result.

The code should write a JSON file to ‘result.json’ with: {“vertices”: [[x,y,z], …], “indices”: [i1, i2, i3, …]}

Parameters:
code : str

Python code that generates a mesh

config : SandboxConfig, optional

Sandbox configuration

Returns:

(vertices, indices)

Return type:

tuple[list[list[float]], list[int]]

Raises:

ValueError – If code fails or doesn’t produce valid mesh data

AI-powered mesh generation using Claude Agent SDK.

Provides an agent that can generate 3D meshes from natural language prompts, optionally processing input data files.

Code execution is sandboxed via Firejail when available, providing: - Filesystem isolation - Network isolation - Resource limits (memory, CPU time) - Capability dropping and seccomp filtering

class ascribe_link.agent_generator.AgentResult(result_type: str | None = None, vertices: list[list[float]] | None = None, indices: list[int] | None = None, normals: list[list[float]] | None = None, volume_shape: list[int] | None = None, volume_dtype: str | None = None, volume_data: str | None = None, volume_spacing: list[float] | None = None, error: str | None = None, submitted: bool = False)[source]

Bases: object

Holds the result from the agent (mesh or volume).

class ascribe_link.agent_generator.TranscriptWriter(path: str | Path, user_prompt: str, model: str | None = None)[source]

Bases: object

Append a human-readable Markdown transcript of an agent run.

Best-effort debugging artifact: every write is wrapped so a transcript failure can never break the agent loop. Messages are identified by duck typing (type name / attributes) rather than SDK imports so this works without claude_agent_sdk installed and with mocked message objects.

record(msg: Any) → None[source]

Append the relevant parts of one SDK message to the transcript.

ascribe_link.agent_generator.create_agent_function(model: str = 'claude-opus-4-8', timeout: float = 3000000.0, sandbox: bool = True, sandbox_config: SandboxConfig | None = None, isolate_process: bool = True)[source]

Create an agent-based generation function for the registry.

Parameters:
model : str

Default Claude model to use.

timeout : float

Default timeout in seconds.

sandbox : bool

If True, wrap Bash commands in Firejail sandbox.

sandbox_config : SandboxConfig, optional

Configuration for sandbox limits.

isolate_process : bool

If True (default), run the agent in a child process so SDK message parsing can never hold this process’s GIL (which stalls the HTTP event loop). Set False to run in-process (tests, debugging).

Returns:

Async function compatible with FunctionRegistry.

Return type:

callable

async ascribe_link.agent_generator.generate_mesh_with_agent(prompt: str, file_path: str | None = None, model: str = 'claude-opus-4-8', timeout: float = 3000000.0, working_dir: str | None = None, sandbox: bool = True, sandbox_config: SandboxConfig | None = None, reporter: ProgressReporter | None = None) → dict[str, Any]

Generate data (mesh, volume, etc.) using an AI agent.

Parameters:
prompt : str

Natural language description of what to generate.

file_path : str, optional

Path to an input data file for the agent to process.

model : str

Claude model to use.

timeout : float

Maximum time in seconds to wait for the agent.

working_dir : str, optional

Working directory for the agent. Defaults to a temp directory.

sandbox : bool

If True, wrap Bash commands in Firejail sandbox.

sandbox_config : SandboxConfig, optional

Configuration for sandbox. Uses defaults if None.

Returns:

Result dictionary with ‘type’ field indicating the data type: - “mesh”: contains vertices, indices - “volume”: contains shape, dtype, data (base64), spacing

Return type:

dict

Raises:
async ascribe_link.agent_generator.generate_with_agent(prompt: str, file_path: str | None = None, model: str = 'claude-opus-4-8', timeout: float = 3000000.0, working_dir: str | None = None, sandbox: bool = True, sandbox_config: SandboxConfig | None = None, reporter: ProgressReporter | None = None) → dict[str, Any][source]

Generate data (mesh, volume, etc.) using an AI agent.

Parameters:
prompt : str

Natural language description of what to generate.

file_path : str, optional

Path to an input data file for the agent to process.

model : str

Claude model to use.

timeout : float

Maximum time in seconds to wait for the agent.

working_dir : str, optional

Working directory for the agent. Defaults to a temp directory.

sandbox : bool

If True, wrap Bash commands in Firejail sandbox.

sandbox_config : SandboxConfig, optional

Configuration for sandbox. Uses defaults if None.

Returns:

Result dictionary with ‘type’ field indicating the data type: - “mesh”: contains vertices, indices - “volume”: contains shape, dtype, data (base64), spacing

Return type:

dict

Raises:
ascribe_link.agent_generator.wrap_agent_output(value: Any)[source]

Coerce an arbitrary agent Python return value into a typed Result.

Dispatch: - VolumeResult / MeshResult -> passthrough - numpy.ndarray (ndim == 3) -> VolumeResult.from_numpy (cast to float32) - pyvista.PolyData / similar -> MeshResult.from_pyvista - anything else -> TypeError

Utilities