diaphora-python 0.1.4

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

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

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)

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

---

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

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

---