# RoboSystems Python Client
[](https://pypi.org/project/robosystems-client/)
[](https://opensource.org/licenses/MIT)
Official Python Client 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-client
```
## Quick Start
```python
from robosystems_client import RoboSystemsSDK
from robosystems_client.api.query import execute_cypher_query
from robosystems_client.models import CypherQueryRequest
# Initialize the SDK
sdk = RoboSystemsSDK(
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=sdk, 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
### 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=sdk,
body=query_request
)
# Get graph analytics and metrics
metrics = await get_graph_metrics.asyncio(
graph_id="your-graph-id",
client=sdk
)
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=sdk,
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(sdk)
# 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(sdk)
# Stream operation events
from robosystems_client.api.operations import stream_operation_events
await stream_operation_events.asyncio(
operation_id="op-123",
client=sdk
)
```
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
## 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=sdk,
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=sdk,
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 Client 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 Client
When the API changes, regenerate the Client from the OpenAPI spec:
```bash
# From localhost (development)
just generate-client http://localhost:8000/openapi.json
# From staging
just generate-client https://staging.api.robosystems.ai/openapi.json
# From production
just generate-client https://api.robosystems.ai/openapi.json
```
### Testing
```bash
just test
just test-cov
```
### Code Quality
```bash
just lint
just format
just typecheck
```
Raw data
{
"_id": null,
"home_page": null,
"name": "robosystems-client",
"maintainer": null,
"docs_url": null,
"requires_python": ">=3.10",
"maintainer_email": null,
"keywords": "api, client, financial, graph, kuzu, robosystems, sdk",
"author": "Harbinger FinLab",
"author_email": null,
"download_url": "https://files.pythonhosted.org/packages/88/09/fbc0a62101c32fc8fc92470a0beb0338839918b53ecbdbdfb8fdbdb90b52/robosystems_client-0.1.9.tar.gz",
"platform": null,
"description": "# RoboSystems Python Client\n\n[](https://pypi.org/project/robosystems-client/)\n[](https://opensource.org/licenses/MIT)\n\nOfficial Python Client 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-client\n```\n\n## Quick Start\n\n```python\nfrom robosystems_client import RoboSystemsSDK\nfrom robosystems_client.api.query import execute_cypher_query\nfrom robosystems_client.models import CypherQueryRequest\n\n# Initialize the SDK\nsdk = RoboSystemsSDK(\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=sdk, 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### 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=sdk, \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=sdk\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=sdk, \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(sdk)\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(sdk)\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=sdk\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## 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=sdk, \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=sdk,\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 Client 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 Client\n\nWhen the API changes, regenerate the Client from the OpenAPI spec:\n\n```bash\n# From localhost (development)\njust generate-client http://localhost:8000/openapi.json\n\n# From staging\njust generate-client https://staging.api.robosystems.ai/openapi.json\n\n# From production\njust generate-client 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",
"bugtrack_url": null,
"license": "MIT",
"summary": "Python Client for RoboSystems financial graph database API",
"version": "0.1.9",
"project_urls": null,
"split_keywords": [
"api",
" client",
" financial",
" graph",
" kuzu",
" robosystems",
" sdk"
],
"urls": [
{
"comment_text": null,
"digests": {
"blake2b_256": "a1789ce7857cf6281a23414ca52a5b71bfedc7c144e04436068eed93926a8975",
"md5": "9c92c888d63766df841e843762229cd8",
"sha256": "b6308a1f5d02ba12876f17984e9cdf89861099c4703e20b478857c4ddd9a4b72"
},
"downloads": -1,
"filename": "robosystems_client-0.1.9-py3-none-any.whl",
"has_sig": false,
"md5_digest": "9c92c888d63766df841e843762229cd8",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.10",
"size": 319204,
"upload_time": "2025-08-11T04:18:07",
"upload_time_iso_8601": "2025-08-11T04:18:07.301053Z",
"url": "https://files.pythonhosted.org/packages/a1/78/9ce7857cf6281a23414ca52a5b71bfedc7c144e04436068eed93926a8975/robosystems_client-0.1.9-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": null,
"digests": {
"blake2b_256": "8809fbc0a62101c32fc8fc92470a0beb0338839918b53ecbdbdfb8fdbdb90b52",
"md5": "7dcf4fe584dedbc53ac6c5b4903751ed",
"sha256": "c3a50095386c0b5599d552b83efd30b5aea5cdd9b4c2531f363d9179676fed03"
},
"downloads": -1,
"filename": "robosystems_client-0.1.9.tar.gz",
"has_sig": false,
"md5_digest": "7dcf4fe584dedbc53ac6c5b4903751ed",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.10",
"size": 124649,
"upload_time": "2025-08-11T04:18:08",
"upload_time_iso_8601": "2025-08-11T04:18:08.581136Z",
"url": "https://files.pythonhosted.org/packages/88/09/fbc0a62101c32fc8fc92470a0beb0338839918b53ecbdbdfb8fdbdb90b52/robosystems_client-0.1.9.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2025-08-11 04:18:08",
"github": false,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"lcname": "robosystems-client"
}
JSON