Diaphora Python SDK

pip install diaphora-python

from diaphora.auth.basic_authenticator import BasicAuthenticator

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

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

from diaphora.auth.authenticator_interface import DiaphoraAuthenticator

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

from diaphora import Diaphora

sdk = Diaphora(authenticator)

from diaphora.models import SearchPlansNamespace

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

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

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

from diaphora.models import SearchResultsStatus

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

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

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"},
)

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)

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

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)

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)

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

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

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,
)

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

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

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)

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

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

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

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

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

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

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

import json

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

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)

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"},
)

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

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

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

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

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

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

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

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

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)