# RoboSystems Python SDK
[](https://pypi.org/project/robosystems-sdk/)
[](https://opensource.org/licenses/MIT)
Official Python SDK for the RoboSystems Financial Knowledge Graph API. Access comprehensive financial data including accounting records, SEC filings, and advanced graph analytics through a type-safe, async-ready Python interface.
## Features
- **Type-safe API client** with full type hints and Pydantic models
- **Async/await support** for high-performance applications
- **Multi-tenant support** with graph-scoped operations
- **Authentication handling** with API key and SSO support
- **Comprehensive error handling** with custom exceptions
- **Pagination support** for large data sets
- **Streaming query support** for memory-efficient processing of large result sets
- **Financial AI Agent** integration for natural language queries
## Installation
```bash
pip install robosystems-sdk
```
## Quick Start
```python
from robosystems_client import RoboSystemsClient
from robosystems_client.api.query import execute_cypher_query
from robosystems_client.models import CypherQueryRequest
# Initialize the client
client = RoboSystemsClient(
base_url="https://api.robosystems.ai",
token="your-api-key",
auth_header_name="X-API-Key",
prefix="" # No prefix needed for API key
)
# Async usage (recommended)
import asyncio
async def main():
# Execute a Cypher query
query = CypherQueryRequest(
query="MATCH (c:Company)-[:HAS_FILING]->(f:Filing) RETURN c.name, f.form_type, f.filing_date LIMIT 10"
)
result = await execute_cypher_query.asyncio(graph_id="your-graph-id", client=client, body=query)
for row in result.data:
print(f"{row['c.name']} filed {row['f.form_type']} on {row['f.filing_date']}")
asyncio.run(main())
```
## Key API Endpoints
### Authentication & User Management
```python
from robosystems_client.api.auth import login_user, get_current_auth_user
from robosystems_client.api.user import create_user_api_key, get_user_graphs
from robosystems_client.models import LoginRequest, CreateAPIKeyRequest
# Login and manage sessions
login_request = LoginRequest(email="user@example.com", password="your-password")
auth_response = await login_user.asyncio(client=client, body=login_request)
# Get current authenticated user
current_user = await get_current_auth_user.asyncio(client=client)
# API key management
api_key_request = CreateAPIKeyRequest(name="My API Key", description="API key for automation")
api_key = await create_user_api_key.asyncio(client=client, body=api_key_request)
# List user's graphs
user_graphs = await get_user_graphs.asyncio(client=client)
```
### Data Connections
```python
from robosystems_client.api.connections import create_connection, sync_connection, list_connections
from robosystems_client.models import CreateConnectionRequest, SyncConnectionRequest
# List existing connections
connections = await list_connections.asyncio(
graph_id="your-graph-id",
client=client
)
# Create a data connection (QuickBooks, bank accounts, etc.)
connection_config = CreateConnectionRequest(
name="QuickBooks Integration",
provider="quickbooks",
config={"company_id": "123456"}
)
connection = await create_connection.asyncio(
graph_id="your-graph-id",
client=client,
body=connection_config
)
# Sync connection data
sync_request = SyncConnectionRequest(
sync_options={"full_sync": True}
)
sync_result = await sync_connection.asyncio(
graph_id="your-graph-id",
connection_id="connection-id",
client=client,
body=sync_request
)
```
### Graph Queries & Analytics
```python
from robosystems_client.api.query import execute_cypher_query
from robosystems_client.api.graph_analytics import get_graph_metrics
from robosystems_client.models import CypherQueryRequest
# Execute Cypher queries with parameters
query_request = CypherQueryRequest(
query="""MATCH (c:Company {ticker: $ticker})-[:HAS_METRIC]->(m:Metric)
WHERE m.fiscal_year >= $start_year
RETURN m.name, m.value, m.fiscal_year
ORDER BY m.fiscal_year DESC""",
parameters={"ticker": "AAPL", "start_year": 2020}
)
results = await execute_cypher_query.asyncio(
graph_id="your-graph-id",
client=client,
body=query_request
)
# Get graph analytics and metrics
metrics = await get_graph_metrics.asyncio(
graph_id="your-graph-id",
client=client
)
print(f"Total nodes: {metrics.total_nodes}")
print(f"Total relationships: {metrics.total_relationships}")
```
### Financial AI Agent
```python
from robosystems_client.api.agent import query_financial_agent
from robosystems_client.models import AgentRequest
# Natural language financial queries
agent_request = AgentRequest(
message="What was Apple's revenue growth over the last 3 years?",
force_extended_analysis=True,
context={"include_schema": True}
)
agent_response = await query_financial_agent.asyncio(
graph_id="your-graph-id",
client=client,
body=agent_request
)
print(f"Response: {agent_response.message}")
```
### Function Patterns
Every API endpoint provides multiple calling patterns:
- **`asyncio()`** - Async call, returns parsed response (recommended)
- **`asyncio_detailed()`** - Async call, returns full Response object
- **`sync()`** - Synchronous call, returns parsed response
- **`sync_detailed()`** - Synchronous call, returns full Response object
## Streaming Support with Extensions
The SDK includes an extensions module with SSE (Server-Sent Events) support for real-time streaming operations:
```python
from robosystems_client.extensions import (
SSEClient,
QueryClient,
OperationClient,
RoboSystemsExtensions
)
from robosystems_client.models import CypherQueryRequest
# Initialize extensions
extensions = RoboSystemsExtensions()
# Use QueryClient for advanced query operations
query_client = QueryClient(client)
# Execute queries with the query client
query = CypherQueryRequest(
query="""MATCH (c:Company)-[:HAS_METRIC]->(m:Metric)
WHERE m.fiscal_year >= 2020
RETURN c.name, m.name, m.value, m.fiscal_year
ORDER BY c.name, m.fiscal_year""",
parameters={}
)
# Monitor long-running operations with SSE
operation_client = OperationClient(client)
# Stream operation events
from robosystems_client.api.operations import stream_operation_events
await stream_operation_events.asyncio(
operation_id="op-123",
client=client
)
```
The extensions module provides:
- SSE client for real-time event streaming
- Query client with advanced query management
- Operation client for monitoring long-running tasks
- Utilities for result processing and caching
## Advanced Features
### Billing & Credit Management
```python
from robosystems_client.api.credits_ import get_credit_summary, check_credit_balance
from robosystems_client.api.billing import get_current_graph_bill, get_graph_usage_details
# Monitor credits and usage
credit_summary = await get_credit_summary.asyncio(client=client)
print(f"Available credits: {credit_summary.available_credits:,}")
print(f"Monthly usage: {credit_summary.used_credits:,}/{credit_summary.total_credits:,}")
# Check credit balance
credit_check = await check_credit_balance.asyncio(
graph_id="your-graph-id",
client=client
)
# Billing information
current_bill = await get_current_graph_bill.asyncio(
graph_id="your-graph-id",
client=client
)
# Detailed usage metrics
usage_details = await get_graph_usage_details.asyncio(
graph_id="your-graph-id",
client=client,
start_date="2024-01-01",
end_date="2024-01-31"
)
```
### MCP Tools Integration
```python
from robosystems_client.api.mcp import list_mcp_tools, call_mcp_tool
from robosystems_client.models import MCPToolCall
# List available MCP tools
tools = await list_mcp_tools.asyncio(client=client)
for tool in tools.tools:
print(f"{tool.name}: {tool.description}")
# Call an MCP tool
tool_call = MCPToolCall(
name="analyze_financial_statement",
arguments={
"company_id": "AAPL",
"statement_type": "income",
"fiscal_year": 2023
}
)
tool_result = await call_mcp_tool.asyncio(
client=client,
body=tool_call
)
print(f"Analysis result: {tool_result.content}")
```
## Error Handling
```python
from robosystems_client.types import Response
from robosystems_client.errors import UnexpectedStatus
import httpx
try:
# API calls that might fail
result = await execute_cypher_query.asyncio(
graph_id="your-graph-id",
client=client,
body=query_request
)
except UnexpectedStatus as e:
# Handle API errors (4xx, 5xx)
print(f"API error: {e.status_code}")
print(f"Error details: {e.content}")
# Parse error response if JSON
if e.status_code == 400:
error_detail = e.content.decode('utf-8')
print(f"Validation error: {error_detail}")
elif e.status_code == 401:
print("Authentication failed - check your API key")
elif e.status_code == 403:
print("Permission denied - check graph access")
elif e.status_code == 429:
print("Rate limit exceeded - retry later")
except httpx.TimeoutException:
print("Request timed out - try again")
except httpx.NetworkError as e:
print(f"Network error: {e}")
except Exception as e:
print(f"Unexpected error: {type(e).__name__}: {e}")
# Using detailed responses for better error handling
from robosystems_client.api.query import execute_cypher_query
response = await execute_cypher_query.asyncio_detailed(
graph_id="your-graph-id",
client=client,
body=query_request
)
if response.status_code == 200:
result = response.parsed
print(f"Success: Query executed successfully")
else:
print(f"Failed with status {response.status_code}")
print(f"Headers: {response.headers}")
print(f"Content: {response.content}")
```
## Development
This SDK is auto-generated from the RoboSystems OpenAPI specification to ensure it stays in sync with the latest API changes.
### Setup
```bash
just venv
just install
```
### Regenerating the SDK
When the API changes, regenerate the SDK from the OpenAPI spec:
```bash
# From localhost (development)
just generate-sdk http://localhost:8000/openapi.json
# From staging
just generate-sdk https://staging.api.robosystems.ai/openapi.json
# From production
just generate-sdk https://api.robosystems.ai/openapi.json
```
### Testing
```bash
just test
just test-cov
```
### Code Quality
```bash
just lint
just format
just typecheck
```
### Publishing
```bash
just build-package
just publish-package
```
Raw data
{
"_id": null,
"home_page": null,
"name": "robosystems-sdk",
"maintainer": null,
"docs_url": null,
"requires_python": ">=3.10",
"maintainer_email": null,
"keywords": "api, client, financial, graph, kuzu, robosystems, sdk",
"author": "RoboSystems Team",
"author_email": null,
"download_url": "https://files.pythonhosted.org/packages/ca/8b/38e1f6bfedaf312ebf87d7c2b12163ce99f5bb7627ec5507f93b4ffec643/robosystems_sdk-0.1.8.tar.gz",
"platform": null,
"description": "# RoboSystems Python SDK\n\n[](https://pypi.org/project/robosystems-sdk/)\n[](https://opensource.org/licenses/MIT)\n\nOfficial Python SDK for the RoboSystems Financial Knowledge Graph API. Access comprehensive financial data including accounting records, SEC filings, and advanced graph analytics through a type-safe, async-ready Python interface.\n\n## Features\n\n- **Type-safe API client** with full type hints and Pydantic models\n- **Async/await support** for high-performance applications \n- **Multi-tenant support** with graph-scoped operations\n- **Authentication handling** with API key and SSO support\n- **Comprehensive error handling** with custom exceptions\n- **Pagination support** for large data sets\n- **Streaming query support** for memory-efficient processing of large result sets\n- **Financial AI Agent** integration for natural language queries\n\n## Installation\n\n```bash\npip install robosystems-sdk\n```\n\n## Quick Start\n\n```python\nfrom robosystems_client import RoboSystemsClient\nfrom robosystems_client.api.query import execute_cypher_query\nfrom robosystems_client.models import CypherQueryRequest\n\n# Initialize the client\nclient = RoboSystemsClient(\n base_url=\"https://api.robosystems.ai\",\n token=\"your-api-key\",\n auth_header_name=\"X-API-Key\",\n prefix=\"\" # No prefix needed for API key\n)\n\n# Async usage (recommended)\nimport asyncio\n\nasync def main():\n # Execute a Cypher query\n query = CypherQueryRequest(\n query=\"MATCH (c:Company)-[:HAS_FILING]->(f:Filing) RETURN c.name, f.form_type, f.filing_date LIMIT 10\"\n )\n result = await execute_cypher_query.asyncio(graph_id=\"your-graph-id\", client=client, body=query)\n \n for row in result.data:\n print(f\"{row['c.name']} filed {row['f.form_type']} on {row['f.filing_date']}\")\n\nasyncio.run(main())\n```\n\n## Key API Endpoints\n\n### Authentication & User Management\n```python\nfrom robosystems_client.api.auth import login_user, get_current_auth_user\nfrom robosystems_client.api.user import create_user_api_key, get_user_graphs\nfrom robosystems_client.models import LoginRequest, CreateAPIKeyRequest\n\n# Login and manage sessions\nlogin_request = LoginRequest(email=\"user@example.com\", password=\"your-password\")\nauth_response = await login_user.asyncio(client=client, body=login_request)\n\n# Get current authenticated user\ncurrent_user = await get_current_auth_user.asyncio(client=client)\n\n# API key management\napi_key_request = CreateAPIKeyRequest(name=\"My API Key\", description=\"API key for automation\")\napi_key = await create_user_api_key.asyncio(client=client, body=api_key_request)\n\n# List user's graphs\nuser_graphs = await get_user_graphs.asyncio(client=client)\n```\n\n### Data Connections\n```python\nfrom robosystems_client.api.connections import create_connection, sync_connection, list_connections\nfrom robosystems_client.models import CreateConnectionRequest, SyncConnectionRequest\n\n# List existing connections\nconnections = await list_connections.asyncio(\n graph_id=\"your-graph-id\",\n client=client\n)\n\n# Create a data connection (QuickBooks, bank accounts, etc.)\nconnection_config = CreateConnectionRequest(\n name=\"QuickBooks Integration\",\n provider=\"quickbooks\",\n config={\"company_id\": \"123456\"}\n)\nconnection = await create_connection.asyncio(\n graph_id=\"your-graph-id\", \n client=client, \n body=connection_config\n)\n\n# Sync connection data\nsync_request = SyncConnectionRequest(\n sync_options={\"full_sync\": True}\n)\nsync_result = await sync_connection.asyncio(\n graph_id=\"your-graph-id\",\n connection_id=\"connection-id\",\n client=client,\n body=sync_request\n)\n```\n\n### Graph Queries & Analytics\n```python\nfrom robosystems_client.api.query import execute_cypher_query\nfrom robosystems_client.api.graph_analytics import get_graph_metrics\nfrom robosystems_client.models import CypherQueryRequest\n\n# Execute Cypher queries with parameters\nquery_request = CypherQueryRequest(\n query=\"\"\"MATCH (c:Company {ticker: $ticker})-[:HAS_METRIC]->(m:Metric)\n WHERE m.fiscal_year >= $start_year\n RETURN m.name, m.value, m.fiscal_year\n ORDER BY m.fiscal_year DESC\"\"\",\n parameters={\"ticker\": \"AAPL\", \"start_year\": 2020}\n)\nresults = await execute_cypher_query.asyncio(\n graph_id=\"your-graph-id\", \n client=client, \n body=query_request\n)\n\n# Get graph analytics and metrics\nmetrics = await get_graph_metrics.asyncio(\n graph_id=\"your-graph-id\", \n client=client\n)\nprint(f\"Total nodes: {metrics.total_nodes}\")\nprint(f\"Total relationships: {metrics.total_relationships}\")\n```\n\n### Financial AI Agent\n```python\nfrom robosystems_client.api.agent import query_financial_agent\nfrom robosystems_client.models import AgentRequest\n\n# Natural language financial queries\nagent_request = AgentRequest(\n message=\"What was Apple's revenue growth over the last 3 years?\",\n force_extended_analysis=True,\n context={\"include_schema\": True}\n)\nagent_response = await query_financial_agent.asyncio(\n graph_id=\"your-graph-id\", \n client=client, \n body=agent_request\n)\nprint(f\"Response: {agent_response.message}\")\n```\n\n### Function Patterns\n\nEvery API endpoint provides multiple calling patterns:\n\n- **`asyncio()`** - Async call, returns parsed response (recommended)\n- **`asyncio_detailed()`** - Async call, returns full Response object \n- **`sync()`** - Synchronous call, returns parsed response\n- **`sync_detailed()`** - Synchronous call, returns full Response object\n\n## Streaming Support with Extensions\n\nThe SDK includes an extensions module with SSE (Server-Sent Events) support for real-time streaming operations:\n\n```python\nfrom robosystems_client.extensions import (\n SSEClient, \n QueryClient, \n OperationClient,\n RoboSystemsExtensions\n)\nfrom robosystems_client.models import CypherQueryRequest\n\n# Initialize extensions\nextensions = RoboSystemsExtensions()\n\n# Use QueryClient for advanced query operations\nquery_client = QueryClient(client)\n\n# Execute queries with the query client\nquery = CypherQueryRequest(\n query=\"\"\"MATCH (c:Company)-[:HAS_METRIC]->(m:Metric)\n WHERE m.fiscal_year >= 2020\n RETURN c.name, m.name, m.value, m.fiscal_year\n ORDER BY c.name, m.fiscal_year\"\"\",\n parameters={}\n)\n\n# Monitor long-running operations with SSE\noperation_client = OperationClient(client)\n\n# Stream operation events\nfrom robosystems_client.api.operations import stream_operation_events\nawait stream_operation_events.asyncio(\n operation_id=\"op-123\",\n client=client\n)\n```\n\nThe extensions module provides:\n- SSE client for real-time event streaming\n- Query client with advanced query management\n- Operation client for monitoring long-running tasks\n- Utilities for result processing and caching\n\n## Advanced Features\n\n### Billing & Credit Management\n```python\nfrom robosystems_client.api.credits_ import get_credit_summary, check_credit_balance\nfrom robosystems_client.api.billing import get_current_graph_bill, get_graph_usage_details\n\n# Monitor credits and usage\ncredit_summary = await get_credit_summary.asyncio(client=client)\nprint(f\"Available credits: {credit_summary.available_credits:,}\")\nprint(f\"Monthly usage: {credit_summary.used_credits:,}/{credit_summary.total_credits:,}\")\n\n# Check credit balance\ncredit_check = await check_credit_balance.asyncio(\n graph_id=\"your-graph-id\", \n client=client\n)\n\n# Billing information\ncurrent_bill = await get_current_graph_bill.asyncio(\n graph_id=\"your-graph-id\", \n client=client\n)\n\n# Detailed usage metrics\nusage_details = await get_graph_usage_details.asyncio(\n graph_id=\"your-graph-id\",\n client=client,\n start_date=\"2024-01-01\",\n end_date=\"2024-01-31\"\n)\n```\n\n### MCP Tools Integration\n```python\nfrom robosystems_client.api.mcp import list_mcp_tools, call_mcp_tool\nfrom robosystems_client.models import MCPToolCall\n\n# List available MCP tools\ntools = await list_mcp_tools.asyncio(client=client)\nfor tool in tools.tools:\n print(f\"{tool.name}: {tool.description}\")\n\n# Call an MCP tool\ntool_call = MCPToolCall(\n name=\"analyze_financial_statement\",\n arguments={\n \"company_id\": \"AAPL\", \n \"statement_type\": \"income\",\n \"fiscal_year\": 2023\n }\n)\ntool_result = await call_mcp_tool.asyncio(\n client=client,\n body=tool_call\n)\nprint(f\"Analysis result: {tool_result.content}\")\n```\n\n## Error Handling\n\n```python\nfrom robosystems_client.types import Response\nfrom robosystems_client.errors import UnexpectedStatus\nimport httpx\n\ntry:\n # API calls that might fail\n result = await execute_cypher_query.asyncio(\n graph_id=\"your-graph-id\", \n client=client, \n body=query_request\n )\nexcept UnexpectedStatus as e:\n # Handle API errors (4xx, 5xx)\n print(f\"API error: {e.status_code}\")\n print(f\"Error details: {e.content}\")\n \n # Parse error response if JSON\n if e.status_code == 400:\n error_detail = e.content.decode('utf-8')\n print(f\"Validation error: {error_detail}\")\n elif e.status_code == 401:\n print(\"Authentication failed - check your API key\")\n elif e.status_code == 403:\n print(\"Permission denied - check graph access\")\n elif e.status_code == 429:\n print(\"Rate limit exceeded - retry later\")\nexcept httpx.TimeoutException:\n print(\"Request timed out - try again\")\nexcept httpx.NetworkError as e:\n print(f\"Network error: {e}\")\nexcept Exception as e:\n print(f\"Unexpected error: {type(e).__name__}: {e}\")\n\n# Using detailed responses for better error handling\nfrom robosystems_client.api.query import execute_cypher_query\n\nresponse = await execute_cypher_query.asyncio_detailed(\n graph_id=\"your-graph-id\",\n client=client,\n body=query_request\n)\n\nif response.status_code == 200:\n result = response.parsed\n print(f\"Success: Query executed successfully\")\nelse:\n print(f\"Failed with status {response.status_code}\")\n print(f\"Headers: {response.headers}\")\n print(f\"Content: {response.content}\")\n```\n\n## Development\n\nThis SDK is auto-generated from the RoboSystems OpenAPI specification to ensure it stays in sync with the latest API changes.\n\n### Setup\n\n```bash\njust venv\njust install\n```\n\n### Regenerating the SDK\n\nWhen the API changes, regenerate the SDK from the OpenAPI spec:\n\n```bash\n# From localhost (development)\njust generate-sdk http://localhost:8000/openapi.json\n\n# From staging\njust generate-sdk https://staging.api.robosystems.ai/openapi.json\n\n# From production\njust generate-sdk https://api.robosystems.ai/openapi.json\n```\n\n### Testing\n\n```bash\njust test\njust test-cov\n```\n\n### Code Quality\n\n```bash\njust lint\njust format\njust typecheck\n```\n\n### Publishing\n\n```bash\njust build-package\njust publish-package\n```\n",
"bugtrack_url": null,
"license": "MIT",
"summary": "Python SDK for RoboSystems financial graph database API",
"version": "0.1.8",
"project_urls": null,
"split_keywords": [
"api",
" client",
" financial",
" graph",
" kuzu",
" robosystems",
" sdk"
],
"urls": [
{
"comment_text": null,
"digests": {
"blake2b_256": "5bf6f55877c360f9e437399dad662d0445255efd9cfe468acecc3d9d31eb2d29",
"md5": "d088d5a4e1e747e87ddc28f5d7293cab",
"sha256": "479ffe6cf731f868142699b5e64a02f8b993087a3eb914ed7a675fc2284417f9"
},
"downloads": -1,
"filename": "robosystems_sdk-0.1.8-py3-none-any.whl",
"has_sig": false,
"md5_digest": "d088d5a4e1e747e87ddc28f5d7293cab",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.10",
"size": 321174,
"upload_time": "2025-08-10T05:12:52",
"upload_time_iso_8601": "2025-08-10T05:12:52.656402Z",
"url": "https://files.pythonhosted.org/packages/5b/f6/f55877c360f9e437399dad662d0445255efd9cfe468acecc3d9d31eb2d29/robosystems_sdk-0.1.8-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": null,
"digests": {
"blake2b_256": "ca8b38e1f6bfedaf312ebf87d7c2b12163ce99f5bb7627ec5507f93b4ffec643",
"md5": "be0a46c92059cf74e9284d4bd08d2f98",
"sha256": "c9b1618ea44f354d72178be850aa464921bf9f46361b68e914d346a32e216aee"
},
"downloads": -1,
"filename": "robosystems_sdk-0.1.8.tar.gz",
"has_sig": false,
"md5_digest": "be0a46c92059cf74e9284d4bd08d2f98",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.10",
"size": 125428,
"upload_time": "2025-08-10T05:12:54",
"upload_time_iso_8601": "2025-08-10T05:12:54.437889Z",
"url": "https://files.pythonhosted.org/packages/ca/8b/38e1f6bfedaf312ebf87d7c2b12163ce99f5bb7627ec5507f93b4ffec643/robosystems_sdk-0.1.8.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2025-08-10 05:12:54",
"github": false,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"lcname": "robosystems-sdk"
}
JSON