flowtasks-sdk 0.2.0

Python SDK for the Flow Intelligence Engine API

Flow SDK for Python

Official Python client for the Flow Intelligence Engine API.

Installation

pip install flowtasks-sdk

Quick Start

from flow_sdk import FlowClient

client = FlowClient(
    base_url="https://flowtasks.io",
    api_key="YOUR_API_KEY",
    organization_id="YOUR_ORG_ID",
)

# Upload a document
with open("contract.pdf", "rb") as f:
    doc = client.upload_document("contract.pdf", f)
print(doc["id"])

# Chat with your documents
answer = client.chat("What are the termination conditions?")
print(answer["answer"])

# Search the knowledge graph
results = client.search_graph("termination clause")
print(results)

# Ask the knowledge graph a question (uses LLM reasoning)
response = client.ask_graph("Which contracts allow termination without cause?")
print(response["answer"])

Async Usage

import asyncio
from flow_sdk import AsyncFlowClient

async def main():
    async with AsyncFlowClient(
        base_url="https://flowtasks.io",
        api_key="YOUR_API_KEY",
        organization_id="YOUR_ORG_ID",
    ) as client:
        answer = await client.chat("Summarize the agreement")
        print(answer)

asyncio.run(main())

Available Methods

Documents

• upload_document(filename, file) -- Upload and ingest a document
• list_documents() -- List all documents in the organization
• get_document(doc_id) -- Get document metadata
• delete_document(doc_id) -- Delete a document

Chat / RAG

• chat(message) -- Send a query and get an AI-generated answer with sources

Knowledge Graph

• search_graph(query) -- Search entities and relationships
• get_graph_stats() -- Get graph statistics (node/edge counts)
• ask_graph(question) -- Ask a natural-language question (LLM-powered)
• reason_graph(question) -- Intent-aware graph reasoning

Legal Analysis

• legal_clause_types() -- List available clause types
• legal_clauses(doc_id) -- Extract clauses from a document
• legal_diff(doc_a, doc_b) -- Compare clauses between two documents
• legal_query(query) -- Query across all clauses

Ontology

• list_domains() -- List built-in domain ontologies
• create_ontology(name, ...) -- Create a custom ontology
• list_custom_ontologies() -- List custom ontologies
• get_resolved_ontology(domain) -- Get merged core + custom ontology

Agent

• agent_run(query) -- Trigger an agent run
• list_agent_runs() -- List past agent runs
• list_risk_signals() -- List detected risk signals

Workflows

• list_workflows() -- List workflows
• create_workflow(name, steps) -- Create a workflow
• run_workflow(workflow_id) -- Execute a workflow

Compliance

• list_compliance_checks() -- List compliance checks
• run_compliance_scan(document_id, framework) -- Run a compliance scan

Webhooks

• list_webhooks() -- List configured webhooks
• create_webhook(url, events) -- Register a webhook

Usage

• get_usage() -- Get current usage and quota status

Error Handling

from flow_sdk import FlowClient, FlowAPIError

client = FlowClient(base_url="https://flowtasks.io", api_key="YOUR_KEY")

try:
    result = client.chat("What is this about?")
except FlowAPIError as e:
    print(f"API error {e.status_code}: {e.detail}")

Authentication

Get your API key by registering at POST /api/v1/auth/register or from the Flow dashboard under Settings.

Pass it as api_key when creating the client. All requests are scoped to your organization via the organization_id parameter.

Links

• API Documentation
• OpenAPI Schema

License

MIT