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.
To install the Rapida Python SDK, use the following command:
pip install rapida-python
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}" ) )
You can configure the Rapida SDK to authenticate using your API Key or Personal Token:
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 ) )
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 ) )
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 aConnectionConfigwith 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", } )
The SDK now ships both the legacy synchronous AgentKitAgent server helpers and
the per-conversation AgentKit V2 runtime.
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()
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), ], ) )
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
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.