Connector Client

The ConnectorClient class provides a Python interface to interact with third-party connectors programmatically.

Overview

The Connector Client allows developers to:

• Connect to third-party services (GitHub, Google Drive, Google Calendar, Google Mail, Chief Retrieval)

• Disconnect from integrated services

• Search connected services using natural language queries

• Handle both streaming and non-streaming responses

Installation

pip install smart-search-sdk

Quick Start

import asyncio
import os
from dotenv import load_dotenv

from smart_search_sdk.connector.client import ConnectorClient
from smart_search_sdk.connector.models import (
    ConnectorRequest,
    ConnectorConnectRequest,
    AppName
)

load_dotenv()

async def main():
    # Initialize the client
    client = ConnectorClient(base_url=os.getenv("SMARTSEARCH_BASE_URL"))

    # Authenticate
    await client.authenticate(token=os.getenv("SMARTSEARCH_TOKEN"))

    # Connect to GitHub
    connect_request = ConnectorConnectRequest(
        callback_url="https://your-app.com/callback"
    )
    response = await client.connect_connector(
        app_name=AppName.GITHUB,
        bosa_token=os.getenv("BOSA_TOKEN", ""),
        request=connect_request
    )
    print(response)

asyncio.run(main())

Class: ConnectorClient

Constructor

ConnectorClient(base_url: str)

Parameters:

• base_url (str): The base URL of the Smart Search API

Example:

import os
from dotenv import load_dotenv

load_dotenv()

client = ConnectorClient(base_url=os.getenv("SMARTSEARCH_BASE_URL"))
await client.authenticate(token=os.getenv("SMARTSEARCH_TOKEN"))

---

Methods

Connect - Connect

Initiate integration with a third-party connector.

Signature

async def connect_connector(
    self,
    app_name: AppName,
    bosa_token: str | None,
    request: ConnectorConnectRequest,
) -> dict

Parameters

Parameter	Type	Required	Description
app_name	AppName	Yes	The connector app to connect to
bosa_token	str | None	No	Optional BOSA token for authentication
request	ConnectorConnectRequest	Yes	Connection request containing callback URL

Returns

dict: Response containing connection status and optional setup URL

Response Structure

{
    "status": "setup_required" | "connected",
    "message": str,
    "setup_url": str  # Optional, only when OAuth is required
}

Examples

Connect to GitHub:

import asyncio
import os
from dotenv import load_dotenv

from smart_search_sdk.connector.client import ConnectorClient
from smart_search_sdk.connector.models import (
    ConnectorConnectRequest,
    AppName
)

load_dotenv()

async def connect_github():
    client = ConnectorClient(base_url=os.getenv("SMARTSEARCH_BASE_URL"))
    await client.authenticate(token=os.getenv("SMARTSEARCH_TOKEN"))

    request = ConnectorConnectRequest(
        callback_url="https://your-app.com/callback"
    )

    response = await client.connect_connector(
        app_name=AppName.GITHUB,
        bosa_token=None,
        request=request
    )

    if response["status"] == "setup_required":
        print(f"Please visit: {response['setup_url']}")
    else:
        print(f"Connected: {response['message']}")

asyncio.run(connect_github())

Connect with BOSA token:

import asyncio
import os
from dotenv import load_dotenv

from smart_search_sdk.connector.client import ConnectorClient
from smart_search_sdk.connector.models import (
    ConnectorConnectRequest,
    AppName
)

load_dotenv()

async def connect_with_token():
    client = ConnectorClient(base_url=os.getenv("SMARTSEARCH_BASE_URL"))
    await client.authenticate(token=os.getenv("SMARTSEARCH_TOKEN"))

    request = ConnectorConnectRequest(
        callback_url="https://your-app.com/callback"
    )

    response = await client.connect_connector(
        app_name=AppName.GOOGLE_DRIVE,
        bosa_token="your-bosa-token",
        request=request
    )

    return response

asyncio.run(connect_with_token())

---

Connect - Disconnect

Remove integration with a third-party connector.

Signature

async def disconnect_connector(
    self,
    app_name: AppName,
    bosa_token: str | None
) -> dict

Parameters

Parameter	Type	Required	Description
app_name	AppName	Yes	The connector app to disconnect from
bosa_token	str | None	No	Optional BOSA token for authentication

Returns

dict: Confirmation of disconnection

Response Structure

{
    "status": "disconnected",
    "message": str
}

Examples

Disconnect from GitHub:

import asyncio
import os
from dotenv import load_dotenv

from smart_search_sdk.connector.client import ConnectorClient
from smart_search_sdk.connector.models import AppName

load_dotenv()

async def disconnect_github():
    client = ConnectorClient(base_url=os.getenv("SMARTSEARCH_BASE_URL"))
    await client.authenticate(token=os.getenv("SMARTSEARCH_TOKEN"))

    response = await client.disconnect_connector(
        app_name=AppName.GITHUB,
        bosa_token=None
    )

    print(f"Disconnected: {response['message']}")

asyncio.run(disconnect_github())

---

Connect - Search

Search a connected third-party service using natural language.

Signature

async def search_connector(
    self,
    app_name: AppName,
    bosa_token: str | None,
    request: ConnectorRequest,
    stream: bool = False,
    timeout: float = TimeoutEnum.SIMPLE_PROCESS_TIMEOUT.value,
) -> dict | AsyncGenerator[StreamYieldResponse | dict, None]

Parameters

Parameter	Type	Required	Default	Description
app_name	AppName	Yes	-	The connector app to search
bosa_token	str | None	No	None	Optional BOSA token
request	ConnectorRequest	Yes	-	Search request with query
stream	bool	No	False	Enable streaming response
timeout	float	No	30.0	Request timeout in seconds

Returns

• Non-streaming: dict with search results

• Streaming: AsyncGenerator yielding response chunks

Response Structure

{
    "data": [
        {
            "content": str,
            "metadata": {
                "title": str,
                "source": str
            },
            "confidence_score": float
        }
    ]
}

Examples

Basic search:

import asyncio
import os
from dotenv import load_dotenv

from smart_search_sdk.connector.client import ConnectorClient
from smart_search_sdk.connector.models import ConnectorRequest, AppName

load_dotenv()

async def search_github():
    client = ConnectorClient(base_url=os.getenv("SMARTSEARCH_BASE_URL"))
    await client.authenticate(token=os.getenv("SMARTSEARCH_TOKEN"))

    request = ConnectorRequest(query="open issues about authentication")

    response = await client.search_connector(
        app_name=AppName.GITHUB,
        bosa_token=None,
        request=request,
        stream=False
    )

    for item in response["data"]:
        print(f"Title: {item['metadata']['title']}")
        print(f"Content: {item['content']}")
        print(f"Score: {item['confidence_score']}")
        print("---")

asyncio.run(search_github())

Streaming search:

import asyncio
import os

from dotenv import load_dotenv

from smart_search_sdk.connector.client import ConnectorClient
from smart_search_sdk.connector.models import ConnectorRequest, AppName

load_dotenv()

async def streaming_search():
    client = ConnectorClient(base_url=os.getenv("SMARTSEARCH_BASE_URL"))
    await client.authenticate(token=os.getenv("SMARTSEARCH_TOKEN"))

    request = ConnectorRequest(query="latest commits")

    stream = await client.search_connector(
        app_name=AppName.GITHUB,
        bosa_token=None,
        request=request,
        stream=True
    )

    async for chunk in stream:
        print(f"Received chunk: {chunk}")

asyncio.run(streaming_search())

Search with timeout:

import asyncio
import os
from dotenv import load_dotenv

from smart_search_sdk.connector.client import ConnectorClient
from smart_search_sdk.connector.models import ConnectorRequest, AppName

load_dotenv()

async def search_with_timeout():
    client = ConnectorClient(base_url=os.getenv("SMARTSEARCH_BASE_URL"))
    await client.authenticate(token=os.getenv("SMARTSEARCH_TOKEN"))

    request = ConnectorRequest(query="documents about Q2 financials")

    response = await client.search_connector(
        app_name=AppName.GOOGLE_DRIVE,
        bosa_token=None,
        request=request,
        stream=False,
        timeout=60.0  # 60 seconds timeout
    )

    return response

asyncio.run(search_with_timeout())

Sample queries by connector:

Google Drive (AppName.GOOGLE_DRIVE):

• "Find all documents about Q2 financial reports"

• "Show me all PDF files from last month"

• "List documents shared with the team"

Google Mail (AppName.GOOGLE_MAIL):

• "Find emails from [e-mail] about project alpha"

• "Show me emails with attachments from this week"

• "Find emails about meeting schedule changes"

Google Calendar (AppName.GOOGLE_CALENDAR):

• "List all my upcoming meetings today"

• "Show me meetings with Alice next week"

• "What meetings are scheduled in the conference room tomorrow?"

GitHub (AppName.GITHUB):

• "Find open issues about authentication"

• "Show me pull requests related to bug fixes"

• "List repositories with recent commits"

---

Models

AppName Enum

Available connector apps:

from smart_search_sdk.connector.models import AppName

AppName.GITHUB           # GitHub integration
AppName.GOOGLE_MAIL      # Gmail integration
AppName.GOOGLE_CALENDAR  # Google Calendar integration
AppName.GOOGLE_DRIVE     # Google Drive integration
AppName.CHIEF_RETRIEVAL  # Chief Retrieval integration

ConnectorConnectRequest

from smart_search_sdk.connector.models import ConnectorConnectRequest

request = ConnectorConnectRequest(
    callback_url="https://your-app.com/oauth/callback"
)

Fields:

• callback_url (str): OAuth callback URL for authentication

ConnectorRequest

from smart_search_sdk.connector.models import ConnectorRequest

request = ConnectorRequest(
    query="your natural language query"
)

Fields:

• query (str): Natural language search query

---

Authentication

The client supports two authentication modes:

1. Default Credentials

Uses Smart Search's internal credentials:

response = await client.connect_connector(
    app_name=AppName.GITHUB,
    bosa_token=None,  # Use default credentials
    request=request
)

2. BOSA Token

Uses provided BOSA token:

response = await client.connect_connector(
    app_name=AppName.GITHUB,
    bosa_token="your-bosa-token",
    request=request
)

---

Error Handling

import asyncio
import os
from dotenv import load_dotenv

from smart_search_sdk.connector.client import ConnectorClient
from smart_search_sdk.connector.models import ConnectorRequest, AppName

load_dotenv()

async def safe_search():
    client = ConnectorClient(base_url=os.getenv("SMARTSEARCH_BASE_URL"))

    try:
        await client.authenticate(token=os.getenv("SMARTSEARCH_TOKEN"))

        request = ConnectorRequest(query="test query")
        response = await client.search_connector(
            app_name=AppName.GITHUB,
            bosa_token=os.getenv("BOSA_TOKEN", ""),
            request=request
        )

        return response

    except Exception as e:
        print(f"Error occurred: {e}")
        # Handle error appropriately
        return None

asyncio.run(safe_search())

---

Complete Examples

Search Multiple Connectors

import asyncio
import os
from dotenv import load_dotenv

from smart_search_sdk.connector.client import ConnectorClient
from smart_search_sdk.connector.models import ConnectorRequest, AppName

load_dotenv()

async def search_all_connectors(query: str):
    client = ConnectorClient(base_url=os.getenv("SMARTSEARCH_BASE_URL"))
    await client.authenticate(token=os.getenv("SMARTSEARCH_TOKEN"))

    connectors = [
        AppName.GITHUB,
        AppName.GOOGLE_DRIVE,
        AppName.GOOGLE_CALENDAR
    ]

    results = {}
    request = ConnectorRequest(query=query)

    for connector in connectors:
        try:
            response = await client.search_connector(
                app_name=connector,
                bosa_token=None,
                request=request
            )
            results[connector.value] = response["data"]
        except Exception as e:
            print(f"Error searching {connector.value}: {e}")

    return results

results = asyncio.run(search_all_connectors("team meetings"))
for connector, data in results.items():
    print(f"\n{connector} results:")
    for item in data[:3]:  # Show top 3 results
        print(f"  - {item['metadata']['title']}")

Setup OAuth Flow

import asyncio
import os
from dotenv import load_dotenv

from smart_search_sdk.connector.client import ConnectorClient
from smart_search_sdk.connector.models import (
    ConnectorConnectRequest,
    AppName
)

load_dotenv()

async def setup_oauth_connector(app_name: AppName):
    client = ConnectorClient(base_url=os.getenv("SMARTSEARCH_BASE_URL"))
    await client.authenticate(token=os.getenv("SMARTSEARCH_TOKEN"))

    request = ConnectorConnectRequest(
        callback_url="https://your-app.com/oauth/callback"
    )

    response = await client.connect_connector(
        app_name=app_name,
        bosa_token=None,
        request=request
    )

    if response["status"] == "setup_required":
        print(f"Authentication required for {app_name.value}")
        print(f"Please visit: {response['setup_url']}")
        print("After authentication, you can start searching.")
    elif response["status"] == "connected":
        print(f"Successfully connected to {app_name.value}")

    return response

asyncio.run(setup_oauth_connector(AppName.GOOGLE_DRIVE))

---

Best Practices

1. Authentication: Always authenticate before making requests

   await client.authenticate(token=os.getenv("SMARTSEARCH_TOKEN"))

2. Resource Management: Use async context managers when possible

   async with ConnectorClient(base_url=url) as client:
       await client.authenticate()
       # Use client

3. Error Handling: Wrap requests in try-except blocks

   try:
       response = await client.search_connector(...)
   except Exception as e:
       logger.error(f"Search failed: {e}")

4. Streaming for Large Results: Use streaming for better performance

   stream = await client.search_connector(..., stream=True)
   async for chunk in stream:
       process_chunk(chunk)

5. Timeout Configuration: Set appropriate timeouts

   response = await client.search_connector(..., timeout=60.0)

6. Secure Tokens: Never hardcode BOSA tokens

   import os
   bosa_token = os.getenv("BOSA_TOKEN")

---

Related Documentation

• CLI Connector Commands

• Web Search Client