Diaphora Python SDK

Python client library for the Diaphora API.

Supported Python versions: 3.11, 3.12, 3.13

Installation

pip install diaphora-python

Authentication

The Diaphora client needs your Diaphora credentials. Pass them directly, or set the DIAPHORA_USERNAME and DIAPHORA_PASSWORD environment variables and call the constructor with no arguments.

Direct credentials:

from diaphora.auth.basic_authenticator import BasicAuthenticator

auth = BasicAuthenticator("user@example.com", "your-password")

Environment variables:

# export DIAPHORA_USERNAME=user@example.com
# export DIAPHORA_PASSWORD=your-password
auth = BasicAuthenticator()

To use a custom auth scheme, implement DiaphoraAuthenticator:

from diaphora.auth.authenticator_interface import DiaphoraAuthenticator

class MyAuthenticator(DiaphoraAuthenticator):
    def token(self) -> str:
        return "my-session-token"

Diaphora

from diaphora import Diaphora

sdk = Diaphora(authenticator)

Quickstart

Search for plans

from diaphora.models import SearchPlansNamespace

# Namespace is one of ALL, DIAPHORA, BARNDOOR, NONE
plans = sdk.store.search_plans(namespace=SearchPlansNamespace.DIAPHORA)

Showing Plan

plan = sdk.store.show_plan(str(plans[0].id))
print(plan.name, plan.description)

Create a plan

from diaphora.models import PlanRequest, PlanRequestVisibility

new_plan = sdk.store.create_plan(PlanRequest(
    name="Daily Summary",
    description="Summarises activity from the past 24 hours",
    visibility=PlanRequestVisibility.ORGANIZATION,
    labels=["summary", "daily"],
    text="<plan definition>",
))

Search results for a plan

from diaphora.models import SearchResultsStatus

results = sdk.store.search_results(
    plan_id=str(plan.id),
    status=SearchResultsStatus.SUCCESS,
)

Run a plan

response = sdk.router.run_plan(
    plan_id=str(plan.id),
    parameters={"animal_type": "feline"},
)

Stream a plan execution

from diaphora.events import START_PHASE, END_PHASE, RESULT_PHASE, StreamEvent

def handle_event(event: StreamEvent):
    if event.event == START_PHASE:
        print(f"[{event.component}] starting {event.session}...")
    elif event.event == END_PHASE:
        print(f"[{event.component}] done")
    elif event.event == RESULT_PHASE:
        print(event.content["document"])

sdk.router.stream_plan(
    plan_id=str(plan.id),
    on_event=handle_event,
    parameters={"animal_type": "feline"},
)

sdk.store

Plans

search_plans

Search plans visible to your organisation. All filters are optional and combinable.

Parameters

Returns list[PlanSummary] | None — each item has .id, .name, .description, .namespace, .labels, .visibility, .parameters, .created_at, .updated_at.

Example

from diaphora.models import SearchPlansNamespace

# All plans
plans = sdk.store.search_plans()

# Filter by namespace and label
plans = sdk.store.search_plans(
    namespace=SearchPlansNamespace.DIAPHORA,
    label=["daily"],
    limit=10,
)
for plan in plans:
    print(plan.name, plan.namespace)

show_plan

Retrieve full details for a single plan, including its definition text and parameter schema.

Parameters

Returns PlanDetails | None — all fields from PlanSummary plus .text (plan definition), .document_template, and .resources.

Example

plan = sdk.store.show_plan(str(plans[0].id))
print(plan.name, plan.description)
print("Parameters:", [p.name for p in plan.parameters])

create_plan

Create a new plan.

Parameters

PlanRequest fields

Returns PlanDetails | None — the newly created plan.

Example

from diaphora.models import PlanRequest, PlanRequestVisibility

new_plan = sdk.store.create_plan(PlanRequest(
    name="Daily Summary",
    description="Summarises activity from the past 24 hours",
    visibility=PlanRequestVisibility.ORGANIZATION,
    labels=["summary", "daily"],
    text="<plan definition>",
))
print(new_plan.id)

update_plan

Overwrite an existing plan's metadata and definition text.

Parameters

Returns PlanDetails | None — the updated plan.

Example

from diaphora.models import PlanRequest, PlanRequestVisibility

updated = sdk.store.update_plan(
    plan_id=str(plan.id),
    body=PlanRequest(
        name="Daily Summary v2",
        description="Updated summary plan",
        visibility=PlanRequestVisibility.ORGANIZATION,
        labels=["summary", "daily", "v2"],
        text="<updated plan definition>",
    ),
)
print(updated.name)

delete_plan

Permanently delete a plan. This action cannot be undone.

Parameters

Returns Response — HTTP 204 on success.

Example

sdk.store.delete_plan(str(plan.id))

list_plan_labels

Return every unique label used across plans in the organisation.

Returns list[str] | None — alphabetically sorted list of label strings.

Example

labels = sdk.store.list_plan_labels()
print(labels)  # ["daily", "reports", "summary"]

Results

search_results

Search execution results with optional filters. All parameters are optional and combinable.

Parameters

Returns list[ResultSummary] | None — each item has .id, .plan_id, .plan_name, .created_at.

Example

from diaphora.models import SearchResultsStatus

# All results for a plan
results = sdk.store.search_results(plan_id=str(plan.id))

# Successful results only, newest 20
recent = sdk.store.search_results(
    plan_id=str(plan.id),
    status=SearchResultsStatus.SUCCESS,
    limit=20,
)

show_results

Retrieve full details for a single result, including the execution output.

Parameters

Returns Response[ResultDetails] — access the parsed object via .parsed. ResultDetails has .id, .plan_id, .plan_name, .created_at, .data (ExecuteResponseBody), and .error.

Example

response = sdk.store.show_results(str(results[0].id))
result = response.parsed
print(result.plan_name, result.created_at)
print(result.data.document)

delete_results

Permanently delete a result record. This action cannot be undone.

Parameters

Returns Response — HTTP 204 on success.

Example

sdk.store.delete_results(str(result.id))

list_results_stats

Return a time-ordered list of result metadata records, optionally bounded by a date range. Useful for building audit logs or activity dashboards.

Parameters

Returns list[Stat] | None — each item has .result_id, .plan_id, .plan_name, .created_at.

Example

from datetime import datetime, timezone

stats = sdk.store.list_results_stats()
for stat in stats:
    print(f"{stat.plan_name} — {stat.created_at}")

# Last 7 days only
from datetime import timedelta
week_ago = datetime.now(timezone.utc) - timedelta(days=7)
recent_stats = sdk.store.list_results_stats(start=week_ago)

Public Links

list_public_links

List all public share links associated with a result.

Parameters

Returns list[PublicLinkSummary] | None — each item has .id, .label, .created_at, .expires_at, .ref_id, .ref_type.

Example

links = sdk.store.list_public_links(str(result.id))
for link in links:
    print(link.label, "expires:", link.expires_at)

create_public_link

Create a time-limited, unauthenticated share link for a result. Anyone with the link ID can fetch the result via show_result_public_link without credentials.

Parameters

PublicLinkRequest fields

Returns PublicLinkSummary | None — the created link, including its .id and .expires_at.

Example

from diaphora.models import PublicLinkRequest

link = sdk.store.create_public_link(
    results_id=str(result.id),
    body=PublicLinkRequest(expires_in="7d", label="Weekly Report"),
)
print(f"Share link ID: {link.id}, expires: {link.expires_at}")

delete_public_link

Revoke a public share link immediately. Any subsequent requests using this link ID will fail.

Parameters

Returns Response — HTTP 204 on success.

Example

sdk.store.delete_public_link(str(result.id), str(link.id))

show_result_public_link

Fetch a result via its public share link. Does not require authentication — useful for embedding results in external applications.

Parameters

Returns Response[ResultDetails] — access the parsed object via .parsed.

Example

response = sdk.store.show_result_public_link(str(link.id))
result = response.parsed
print(result.data.document)

Tools & Schema

list_tools

List all tool sets available in the organisation.

Returns list[ToolsSummary] | None — each item has .id, .name, .default.

Example

tools = sdk.store.list_tools()
for t in tools:
    print(t.name, "(default)" if t.default else "")

show_tools

Get full details of a tool set, including its MCP servers, API CP servers, and collections.

Parameters

Returns Response[ToolsDetails] — access via .parsed. ToolsDetails has .id, .name, .default, .mcp_servers, .api_cps, .collections.

Example

response = sdk.store.show_tools(str(tools[0].id))
details = response.parsed
print(f"{details.name} — {len(details.mcp_servers)} MCP servers")

show_default_tools

Get the organisation's default tool set.

Returns Response[ToolsDetails] — access via .parsed.

Example

response = sdk.store.show_default_tools()
default = response.parsed
if default:
    print(f"Default tool set: {default.name}")

get_frags_schema

Retrieve the Frags JSON schema that describes the plan definition format. Useful for validating plan text before submission.

Returns Response — the raw JSON schema is in .content.

Example

import json

response = sdk.store.get_frags_schema()
schema = json.loads(response.content)

sdk.router

Plan Execution

stream_plan delivers StreamEvent objects to the callback:

run_plan

Execute a plan synchronously and block until the result is ready.

Parameters

Returns ExecuteResponseBody with fields:

Example

response = sdk.router.run_plan(
    plan_id=str(plan.id),
    parameters={"animal_type": "feline"},
)
print(response.result)
if response.document:
    print(response.document)
if response.warnings:
    print("Warnings:", response.warnings)

stream_plan

Execute a plan and receive progress events in real time via Server-Sent Events. The optional on_event callback is invoked for each event as it arrives. Returns the final result content when the stream ends.

Parameters

Returns dict | None — the content dict from the RESULT_PHASE event, or None if the stream ended without a result. content["document"] holds the rendered output text.

Example

from diaphora.events import START_PHASE, END_PHASE, RESULT_PHASE, StreamEvent

content = sdk.router.stream_plan(
    plan_id=str(plan.id),
    on_event=handle_event,
    parameters={"animal_type": "feline"},
)

MCP Tools

check_plan_mcp_requirements

List the MCP servers required by a plan and whether each one is currently authenticated.

Parameters

Returns list[McpRequirement] | None — each item has .name, .status (ready or not_ready), .url (the authentication URL when not_ready), .authentication_method.

Example

requirements = sdk.router.check_plan_mcp_requirements(str(plan.id))
for req in requirements:
    print(f"{req.name}: {req.status}")
    if req.status == "not_ready":
        print(f"  Authenticate at: {req.url}")

refresh_plan_mcp_requirements

Force re-evaluation of which MCP servers a plan needs and refresh their authentication status. Call this after completing an OAuth flow to confirm that requirements are now satisfied.

Parameters

Returns list[McpServer] | None — the updated list of MCP servers for the plan, each with .id, .name, .url, .authentication_method.

Example

servers = sdk.router.refresh_plan_mcp_requirements(str(plan.id))
for server in servers:
    print(f"{server.name}: {server.url}")

check_tool_mcp_requirements

List MCP authentication requirements across all tool sets in the organisation. Equivalent to check_plan_mcp_requirements but not scoped to a single plan.

Returns list[McpRequirement] | None

Example

reqs = sdk.router.check_tool_mcp_requirements()
if reqs:
    not_ready = [r for r in reqs if r.status == "not_ready"]
    print(f"{len(not_ready)} servers need authentication")

list_tool_commands

List the commands exposed by a specific MCP server within a tool set.

Parameters

Returns list[McpCommand] | None — each item describes a callable command on the MCP server.

Example

commands = sdk.router.list_tool_commands(
    tools_id=str(tools[0].id),
    server_id=str(server.id),
)
for cmd in commands:
    print(cmd.name)

call_tool_command

Execute a single command on an MCP server.

Parameters

Returns CallToolCommandResponse200 | None

Example

result = sdk.router.call_tool_command(
    tools_id=str(tools[0].id),
    server_id=str(server.id),
    command_name="list_files",
)

list_mcp_auth_cache

List all cached MCP OAuth tokens for the current user.

Returns list[AuthCache] | None — each item has .id, .domain, .created_at, .expiry, and optionally .client_id.

Example

cache = sdk.router.list_mcp_auth_cache()
for entry in cache:
    print(f"{entry.domain} — expires {entry.expiry}")

delete_mcp_auth_cache

Revoke a cached MCP OAuth token. The next request to that MCP server will require re-authentication.

Parameters

Returns None

Example

cache = sdk.router.list_mcp_auth_cache()
sdk.router.delete_mcp_auth_cache(str(cache[0].id))

mcp_callback

Handle an OAuth redirect callback from an MCP server after the user completes authentication. Typically called from a redirect endpoint in your web application.

Parameters

Returns Response

Example

# In a web framework route handler:
# GET /mcp/callback?state=...&code=...
sdk.router.mcp_callback(state=request.args["state"], code=request.args["code"])  

render_template

Render a Diaphora template string against a given scope.

Parameters

RenderTemplate fields

Returns Response — the rendered output is in .content.

Example

from diaphora.models import RenderTemplate
from diaphora.services.frags_router_open_api_client.models import RenderTemplateScope

scope = RenderTemplateScope()
scope["name"] = "Rayna"
response = sdk.router.render_template(
    RenderTemplate(template="Hello {{name}}", scope=scope)
)
print(response.content)