Tool Servers

Where you turn Python code into Convilyn tools.

ToolServer

from convilyn_sdk import ToolServer

server = ToolServer(
    name="my-analyzer",
    description="Document analysis tools",
    version="0.1.0",
)

The developer-facing entry point. Owns a tool registry, an HTTP entrypoint (via server.run()), and the data-store handle that tools can use to stash large payloads off the LLM context.

Constructor parameters

Registering a tool

@server.tool(description="Analyze text length and word count")
async def analyze_text(text: str, language: str = "en") -> dict:
    return {"length": len(text), "words": len(text.split()), "lang": language}

The decorator derives JSON Schema for the tool from the Python type hints — str, int, float, bool, dict, list, Optional[T], T | None, list[str], dict[str, Any], Literal[...], Enum subclasses, and Pydantic BaseModel subclasses are all supported.

If your tool needs platform context (a request ID, a place to store large data, progress reporting), accept a ToolContext parameter — the SDK injects it without exposing it on the wire schema:

from convilyn_sdk import ToolContext

@server.tool(description="Process and stash full result")
async def process(ctx: ToolContext, text: str) -> dict:
    full_result = expensive_op(text)
    ref_id = await ctx.data_store.store(full_result)
    ctx.report_progress(80, "Stored full result")
    return {"ref_id": ref_id, "summary": f"Processed {len(text)} chars"}

Running the server

if __name__ == "__main__":
    server.run()

server.run() boots the FastAPI/uvicorn process locally. Hosting targets (Lambda, Fargate, VM) are documented in the package's docs/DEPLOYMENT.md.

ConvilynServer

A thin re-export alias of ToolServer. Prefer ToolServer in new code; ConvilynServer is kept for symmetry with ConvilynClient and the historical "Convilyn-prefixed" naming.

ToolContext

Injected into any tool function that declares it as a parameter. Read-only handle to per-request platform state.

class ToolContext:
    @property
    def request_id(self) -> str: ...

    @property
    def data_store(self) -> DataStoreProtocol: ...

    def report_progress(self, percent: int, message: str) -> None: ...

ToolContext is never serialised to JSON Schema or exposed to the LLM — the SDK strips it from the tool signature before publishing the wire spec.

ConvilynManifest

The build artefact that lists every server + every tool in a deployable bundle.

from convilyn_sdk import ConvilynManifest

manifest = server.to_manifest()           # build from a ToolServer
manifest.save("convilyn.manifest.json")   # writes JSON to disk
loaded = ConvilynManifest.load("convilyn.manifest.json")

Methods:

• to_json(indent: int = 2) -> str
• to_dict() -> dict[str, Any]
• save(path: str | Path = "convilyn.manifest.json") -> Path
• ConvilynManifest.load(path: str | Path = "convilyn.manifest.json") -> ConvilynManifest
• ConvilynManifest.from_json(data: str) -> ConvilynManifest

The manifest is the source of truth for catalog publishes; deployment scripts (convilyn-author deploy) read it to know which artefacts to push.

Data stores

Tools return small summaries; full payloads go into a data store so the LLM context stays focused.

InMemoryDataStore

The default — process-local, fine for unit tests and single-process dev runs.

from convilyn_sdk import InMemoryDataStore

store = InMemoryDataStore()
ref_id = await store.store({"big": "payload"})
data = await store.get(ref_id)

DataStoreProtocol

The interface every store implements. Authors with custom storage (DynamoDB, Redis, S3) implement this Protocol:

class DataStoreProtocol(Protocol):
    async def store(self, data: dict[str, Any]) -> str: ...
    async def get(self, ref_id: str) -> dict[str, Any] | None: ...

A built-in DynamoDataStore is also available for AWS deployments; pass it as server = ToolServer(..., config=SDKConfig(data_store=...)).

ToolCatalog

Read-only view of an already-published catalog — useful for convilyn-author CLI commands and any inspection script that wants to enumerate what's deployed.

from convilyn_sdk import ToolCatalog

catalog = ToolCatalog("convilyn.manifest.json")
catalog.list_servers()                    # → list[ServerInfo]
catalog.list_tools(server="my-analyzer")  # → list[ToolInfo]
catalog.describe("my-analyzer__analyze_text")  # → ToolInfo | None
catalog.search("text")                    # → list[ToolInfo] matching the query

ConvilynClient

Async HTTP client for the catalog / server publish endpoints. Used by the convilyn-author CLI under the hood; available if you need to script publishes from your own code:

from convilyn_sdk import ConvilynClient

async with ConvilynClient(api_key="ck_...") as client:
    await client.submit_server(manifest)
    servers = await client.list_servers()

Errors surface as ConvilynClientError(status_code: int, detail: str).

Configuration

SDKConfig reads the same env vars the consumer SDK uses (CONVILYN_API_KEY, CONVILYN_BASE_URL), plus a few server-side knobs:

• CONVILYN_INBOUND_SECRET — HMAC secret for verifying signed requests from the platform (recommended in prod; optional in dev)
• CONVILYN_DATA_STORE — "memory" (default) or "dynamo" for a managed AWS deployment
• CONVILYN_PROGRESS_BACKEND — "console" (default) emits progress to stderr; the platform supplies its own backend at runtime

See policies for failure / retry / fallback knobs that apply once a tool is wired into a workflow.