flow-platform-sdk 3.0.0

Python SDK for Uniphore Flow Platform APIs

Flow Platform SDK

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

Usage

Call any backend tool by name through the tools.* proxy:

import asyncio
from flow_platform_sdk import tools

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

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

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

asyncio.run(main())

tools.<name>(**kwargs) dispatches to the backend tool registered under that name -- platform API, connector, or spec action. New backend tools are callable immediately with no SDK code change and no SDK release.

API Reference

Data Connectors

# List all connectors
await tools.list_connectors()

# Inspect schema -- table names only (cheap)
await tools.inspect_schema(connector_id="conn_123", detail="tables")

# Inspect schema -- columns per table (default), optionally filtered
await tools.inspect_schema(
    connector_id="conn_123",
    detail="columns",
    table_filter="customer*",
)

# Execute a read-only query (also how you fetch sample rows)
await tools.execute_query(
    connector_id="conn_123",
    sql_query="SELECT * FROM customers LIMIT 5",
    max_rows=1000,
)

Knowledge Base

# List knowledge bases
await tools.list_knowledge_bases()

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

Agent Evaluation

# Invoke agent with draft skill
result = await tools.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
)

Deprecated: platform_api.*

platform_api.* is the legacy fixed-op surface (platform_api.list_connectors(), etc.). It still works but emits a DeprecationWarning on use and will be removed in a future major version.

It is deprecated because every backend tool change forces a matching SDK code change and a semver-breaking release (e.g. collapsing four schema methods into inspect_schema required a major bump). The tools.* proxy routes any tool by name, so the backend can add, change, or remove tools without an SDK release.

Migrate by replacing platform_api.X(...) with tools.X(...) -- the method names and keyword arguments are unchanged.

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 .