Rapida Python SDK

The Rapida Python SDK provides a powerful interface for interacting with Rapida AI services. This SDK simplifies the process of making API calls, handling authentication, and managing responses from Rapida endpoints.

Installation

To install the Rapida Python SDK, use the following command:

pip install rapida-python

Quick Start

Here's how to get started with the Rapida Python SDK:

from rapida import ConnectionConfig

connection_config = ConnectionConfig.default_connection_config(
    ConnectionConfig.with_sdk(
        "{your-api-key-here}"
    )
)

Authentication

You can configure the Rapida SDK to authenticate using your API Key or Personal Token:

Authenticating with API Key

import os
from rapida.connections import ConnectionConfig

connection_config = ConnectionConfig.default_connection_config(
    ConnectionConfig.with_sdk(
        os.environ["RAPIDA_API_KEY"]  # API Key from environment variables
    )
)

Authenticating with Personal Token

import os
from rapida.connections import ConnectionConfig

connection_config = ConnectionConfig.default_connection_config(
    ConnectionConfig.with_personal_token(
        os.environ["RAPIDA_AUTHORIZATION_TOKEN"],  # Authorization Token
        os.environ["RAPIDA_AUTH_ID"],              # Authentication ID
        os.environ["RAPIDA_PROJECT_ID"],           # Project ID
    )
)

Configuration Options

ConnectionConfig accepts multiple options for configuring the SDK:

• with_sdk(api_key: str): Sets the API key for authentication.
• with_personal_token(auth_token: str, auth_id: str, project_id: str): Configures the connection for personal tokens.
• with_webplugin_client(api_key: str, user_id: Optional[str] = None): Configures web plugin client authentication.
• with_debugger(authorization: str, user_id: str, project_id: str): Configures debugger authentication.
• with_custom_endpoint(endpoint: Optional[dict] = None, debug: Optional[bool] = None): Overrides the default assistant, web, and endpoint API hosts.
• with_local(): Uses local service endpoints and insecure gRPC channels for local development.
• default_connection_config(auth): Creates a ConnectionConfig with the supplied auth metadata.

Example using custom endpoints:

from rapida import ConnectionConfig

connection_config = ConnectionConfig.default_connection_config(
    ConnectionConfig.with_sdk("{your-api-key-here}")
).with_custom_endpoint(
    {
        "assistant": "assistant.example.com:50051",
        "web": "api.example.com:50051",
        "endpoint": "endpoint.example.com:50051",
    }
)

AgentKit

The SDK now ships both the legacy synchronous AgentKitAgent server helpers and the per-conversation AgentKit V2 runtime.

Legacy AgentKit Server

AgentKitServer keeps the existing synchronous gRPC server model while adding:

• ordered middleware for incoming AgentKit calls
• standard gRPC health on /grpc.health.v1.Health/Check
• optional HTTP health on a separate host and port
• existing TLS and auth compatibility helpers

from rapida import AgentKitAgent, AgentKitServer


class LegacyAgent(AgentKitAgent):
    def Talk(self, request_iterator, context):
        for request in request_iterator:
            if self.is_initialization_request(request):
                yield self.initialization_response(request.initialization)
            elif self.is_text_message(request):
                yield self.assistant_response(
                    self.get_message_id(request),
                    "hello from legacy agent",
                    completed=True,
                )


server = AgentKitServer(
    agent=LegacyAgent(),
    host="0.0.0.0",
    port=50051,
    middleware=[
        lambda context: context.metadata_value("authorization") == "secret"
    ],
    http_health_check={"host": "0.0.0.0", "port": 8080, "path": "/healthz"},
)
server.start()
server.wait_for_termination()

AgentKit V2

Agent.runner(...) creates one isolated Agent instance per conversation stream. Each conversation gets its own mutable state, packet helpers, control helpers, and observability helpers.

from rapida import Agent, AgentKitServer


class GreetingAgent(Agent):
    def on_user(self, user):
        self.reply(f"hello {user.id}")
        self.metric({"name": "agent.user_turns", "value": 1})


server = AgentKitServer(
    agent=Agent.runner(GreetingAgent),
    host="0.0.0.0",
    port=50051,
)
server.start()
server.wait_for_termination()

Route by assistant ID and version when one process hosts more than one agent:

from rapida import Agent, AgentRoute, AgentRunnerOptions


class SalesAgent(Agent):
    def on_user(self, user):
        self.reply("sales")


class SupportAgent(Agent):
    def on_user(self, user):
        self.reply("support")


runner = Agent.runner(
    AgentRunnerOptions(
        default=SupportAgent,
        agents=[
            AgentRoute(assistant_id=101, version="v2", agent=SalesAgent),
            AgentRoute(assistant_id=101, agent=SupportAgent),
        ],
    )
)

Compatibility

This SDK requires Python 3.9 or later. Ensure your system meets this requirement:

python --version

To upgrade or specify a version, use the following command:

pip install --upgrade rapida-python

---

Conclusion

The Rapida Python SDK provides everything necessary to integrate seamlessly with Rapida AI services, offering flexible configuration and authentication options. With the examples provided, you should be able to get started quickly and make advanced API calls as needed.