flow-platform-sdk 2.5.0

Python SDK for Uniphore Flow Platform APIs

Flow Platform SDK

Python SDK for interacting with Uniphore Flow Platform services via async RPC.

Usage

import asyncio
from flow_platform_sdk import platform_api

async def main():
    # List data connectors
    connectors = await platform_api.list_connectors()

    # Execute SQL query
    result = await platform_api.execute_query(
        connector_id="conn_123",
        sql_query="SELECT * FROM customers LIMIT 10",
        max_rows=10
    )

    # Query knowledge base
    answer = await platform_api.query_knowledge_base(
        knowledge_base_id="kb_123",
        query="What is the revenue?"
    )

asyncio.run(main())

API Reference

Data Connectors

# List all connectors
await platform_api.list_connectors()

# Get schema
await platform_api.get_schema(connector_id="conn_123")

# Execute query
await platform_api.execute_query(
    connector_id="conn_123",
    sql_query="SELECT * FROM table",
    max_rows=1000
)

# Discover schema
await platform_api.discover_schema(
    connector_id="conn_123",
    include_sample_data=True,
    table_filter="customer*"
)

# Get table info
await platform_api.get_table_info(
    connector_id="conn_123",
    table_name="customers"
)

# Get sample data
await platform_api.get_sample_data(
    connector_id="conn_123",
    table_name="customers",
    limit=5
)

Knowledge Base

# List knowledge bases
await platform_api.list_knowledge_bases()

# Query knowledge base
await platform_api.query_knowledge_base(
    knowledge_base_id="kb_123",
    query="What are the main features?"
)

Agent Evaluation

# Invoke agent with draft skill
result = await platform_api.invoke_agent(
    agent_spec_id="agent_123",
    draft_skill_id="draft_456",
    input_message="What is the weather today?",
    context={"location": "San Francisco"},
    timeout=60
)

Architecture

The SDK uses stdio-based RPC for communication with backend services:

• Requests sent to stdout: REMOTE_SERVICE_CALL:{...}
• Responses read from stdin: REMOTE_TOOL_RESULT:{...}

Designed for sandboxed execution environments.

Configuration

• Default timeout: 120 seconds per request
• Daemon threads for clean exit on timeout

Development

# Setup
uv sync

# Format code
uvx ruff format .
uvx ruff check .