diaphora-python 1.0.7

Python SDK for the Diaphora API

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. You can either pass these directly to the constructor or via environment variables.

from diaphora.auth.basic_authenticator import BasicAuthenticator

auth = BasicAuthenticator("[e-mail]", "your-password")

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)

Attribute	Description
sdk.store	Plan and result management
sdk.router	Plan execution and MCP tools

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

handle_event is a function that can be defined with a custom implementation. The function here is just an example function for printing the contents of the streamed data

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

Method	Description
search_plans(label, search, namespace, limit, offset)	Search plans. namespace is one of ALL, DIAPHORA, BARNDOOR, NONE
show_plan(plan_id)	Get plan details
create_plan(body: PlanRequest)	Create a plan
update_plan(plan_id, body: PlanRequest)	Update a plan
delete_plan(plan_id)	Delete a plan
list_plan_labels()	List all labels used across plans

---

search_plans

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

Parameters

Parameter	Type	Required	Description
label	list[str]	No	Filter to plans that have all of the given labels
search	str	No	Full-text search across plan names and descriptions
namespace	SearchPlansNamespace	No	Filter by namespace — ALL, DIAPHORA, BARNDOOR, or NONE. Defaults to ALL
limit	int	No	Maximum number of results to return
offset	int	No	Number of results to skip (for pagination)

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

Parameter	Type	Required	Description
plan_id	str	Yes	ID of the plan to retrieve

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

Parameter	Type	Required	Description
body	PlanRequest	Yes	Plan definition — see fields below

PlanRequest fields

Field	Type	Required	Description
name	str	Yes	Display name for the plan
description	str	Yes	Human-readable summary
visibility	PlanRequestVisibility	Yes	NAMESPACE, ORGANIZATION, or USER
labels	list[str]	Yes	Tags for grouping and filtering
text	str	Yes	Plan definition body
namespace	str	No	Namespace to assign the plan to
document_template	str	No	Template string for rendering the result document

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

Parameter	Type	Required	Description
plan_id	str	Yes	ID of the plan to update
body	PlanRequest	Yes	Updated plan definition — same fields as create_plan

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

Parameter	Type	Required	Description
plan_id	str	Yes	ID of the plan to delete

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

Method	Description
search_results(search, status, limit, offset, start, end, plan_id)	Search results. status is SUCCESS or ERROR
show_results(results_id)	Get result details
delete_results(results_id)	Delete a result
list_results_stats(start, end)	Get result statistics

---

search_results

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

Parameters

Parameter	Type	Required	Description
search	str	No	Full-text search across result data
status	SearchResultsStatus	No	Filter by outcome — SUCCESS or ERROR
limit	int	No	Maximum number of results to return
offset	int	No	Number of results to skip (for pagination)
start	datetime	No	Return only results created at or after this time
end	datetime	No	Return only results created at or before this time
plan_id	str	No	Filter to results from a specific plan

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

Parameter	Type	Required	Description
results_id	str	Yes	ID of the result to retrieve

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

Parameter	Type	Required	Description
results_id	str	Yes	ID of the result to delete

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

Parameter	Type	Required	Description
start	datetime	No	Return only stats at or after this time
end	datetime	No	Return only stats at or before this time

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

Method	Description
list_public_links(results_id)	List public links for a result
create_public_link(results_id, body: PublicLinkRequest)	Create a public link. expires_in accepts values like "7d" or "24h"
delete_public_link(results_id, public_link_id)	Delete a public link
show_result_public_link(public_link_id)	Fetch a result via public link (no auth required)

---

list_public_links

List all public share links associated with a result.

Parameters

Parameter	Type	Required	Description
results_id	str	Yes	ID of the result whose links to list

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

Parameter	Type	Required	Description
results_id	str	Yes	ID of the result to share
body	PublicLinkRequest	Yes	Link configuration — see fields below

PublicLinkRequest fields

Field	Type	Required	Description
expires_in	str	Yes	How long until the link expires, e.g. "7d", "24h", "30m"
label	str	Yes	Human-readable name for the link

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

Parameter	Type	Required	Description
results_id	str	Yes	ID of the result the link belongs to
public_link_id	str	Yes	ID of the link to revoke

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

Parameter	Type	Required	Description
public_link_id	str	Yes	ID of the public link

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

Method	Description
list_tools()	List available tools
show_tools(tool_id)	Get tool details
show_default_tools()	Get default tools
get_frags_schema()	Get the Frags JSON 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

Parameter	Type	Required	Description
tool_id	str	Yes	ID of the tool set to retrieve

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

Method	Description
run_plan(plan_id, parameters)	Execute a plan synchronously
stream_plan(plan_id, on_event, parameters)	Execute a plan and receive SSE events via callback

stream_plan delivers StreamEvent objects to the callback:

event.event	Additional fields	Description
START_PHASE	component, session	A component started
END_PHASE	component	A component finished
RESULT_PHASE	content	Result payload — content["document"] holds the output text

---

run_plan

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

Parameters

Parameter	Type	Required	Description
plan_id	str	Yes	ID of the plan to execute
parameters	dict	No	Key/value pairs matching the plan's parameter definitions

Returns ExecuteResponseBody with fields:

Field	Type	Description
.result	ExecuteResponseBodyResult	Structured output produced by the plan — call .to_dict() for a plain dict
.document	str | None	Rendered document string, if the plan has a document template
.warnings	list[str]	Non-fatal warnings emitted during execution

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

Parameter	Type	Required	Description
plan_id	str	Yes	ID of the plan to execute
on_event	Callable[[StreamEvent], None]	No	Callback invoked for each SSE event
parameters	dict	No	Key/value pairs matching the plan's parameter definitions

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

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"])

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

---

MCP Tools

Method	Description
check_plan_mcp_requirements(plan_id)	Check which MCP servers a plan needs and their auth status
refresh_plan_mcp_requirements(plan_id)	Force-refresh MCP requirement status
check_tool_mcp_requirements()	Check global MCP requirements
list_tool_commands(tools_id, server_id)	List commands on an MCP server
call_tool_command(tools_id, server_id, command_name)	Execute an MCP command
list_mcp_auth_cache()	List cached MCP OAuth tokens
delete_mcp_auth_cache(cache_id)	Revoke a cached MCP token
mcp_callback(state, code)	Handle an OAuth redirect callback
render_template(body: RenderTemplate)	Render a template

---

check_plan_mcp_requirements

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

Parameters

Parameter	Type	Required	Description
plan_id	str	Yes	ID of the plan to inspect

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

Parameter	Type	Required	Description
plan_id	str	Yes	ID of the plan whose requirements to refresh

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

Parameter	Type	Required	Description
tools_id	str	Yes	ID of the tool set
server_id	str	Yes	ID of the MCP server within that tool set

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

Parameter	Type	Required	Description
tools_id	str	Yes	ID of the tool set
server_id	str	Yes	ID of the MCP server
command_name	str	Yes	Name of the command to execute

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

Parameter	Type	Required	Description
cache_id	str	Yes	ID of the cache entry to revoke

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

Parameter	Type	Required	Description
state	str	Yes	The state query parameter from the OAuth redirect
code	str	Yes	The code query parameter from the OAuth redirect

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

Parameter	Type	Required	Description
body	RenderTemplate	Yes	Template and scope — see fields below

RenderTemplate fields

Field	Type	Required	Description
template	str	Yes	The template string to render
scope	RenderTemplateScope	Yes	Scope context — construct with RenderTemplateScope() and set attributes via dictionary access

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"] = "World"
response = sdk.router.render_template(
    RenderTemplate(template="Hello {{name}}", scope=scope)
)
print(response.content)

---