PinionAI Python Library
This is the official Python client library for the PinionAI platform. It provides a convenient, asynchronous way to interact with PinionAI agents, manage sessions, and use its various features including AI interactions and gRPC messaging.
PinionAI Python Package: PyPi Python
Installation
From PyPI
This package is available on PyPI and can be installed with pip or uv. We recommend uv for its speed.
With uv
If you don't have uv, you can install it from astral.sh.
Once uv is installed, you can install the pinionai package from PyPI:
With pip
If you prefer to use pip, you can still install the package with:
From GitHub
To install the latest development version directly from the GitHub repository:
Optional Features
The base install covers the common client and session workflow. Optional integrations are grouped into extras so you can keep the install lean and only pull in the packages you need.
- gcp: Google Cloud Storage support (google-cloud-storage)
- aws: AWS S3 support (boto3)
- openai: Support for OpenAI models (openai)
- anthropic: Support for Anthropic models (anthropic)
- javascript: Support for running JavaScript snippets (mini-racer)
- sendgrid: Support for running SendGrid delivery
- twilio: Support for SMS delivery
- mcp: FastMCP tool integration (fastmcp)
- data: Tabular and CSV helpers (pandas)
- postgres: PostgreSQL connector support (psycopg)
- google: Google Gemini / Vertex AI support (google-genai)
To install one or more optional features, specify them in brackets. For example, to get support for GCP, AWS, and Google AI:
To install all optional features at once, use the all extra:
Options include:
- dev = [ "build", "twine", "ruff", "grpcio-tools", ]
- gcp = ["google-cloud-storage"]
- aws = ["boto3"]
- openai = ["openai"]
- anthropic = ["anthropic"]
- javascript = ["mini-racer"]
- sendgrid = ["sendgrid"]
- twilio = ["twilio"]
- mcp = ["fastmcp"]
- data = ["pandas"]
- postgres = ["psycopg[binary,pool]"]
- google = ["google-genai>=1.72.0"]
- all = [ "pinionai[gcp,aws,openai,anthropic,javascript,twilio,sendgrid,mcp,data,postgres,google]" ]
Adding to Requirements
To add this library to your project's requirements file, you can use the following formats.
For requirements.txt or requirements.in:
# For a specific version from PyPI
pinionai==0.3.1
# With optional features
pinionai[gcp,openai]==0.3.1
# From the main branch on GitHub
git+https://github.com/pinionai/pinionai-package.git@main
Usage
Here's a complete, fully functional example of how to use the AsyncPinionAIClient. In the following complete example, we run a Streamlit chat, OR you can run it as a CLI app.
PinionAI Agent Github: Pre-built Agent Project
import streamlit as st
import os
import time
import asyncio
from io import StringIO
from pinionai import AsyncPinionAIClient
from pinionai.exceptions import PinionAIConfigurationError, PinionAIError
import threading
from dotenv import load_dotenv
load_dotenv()
def run_coroutine_in_event_loop(coroutine):
"""Runs a coroutine in the app's persistent event loop."""
loop = get_event_loop()
return asyncio.run_coroutine_threadsafe(coroutine, loop).result()
def get_event_loop():
"""Gets or creates the app's persistent event loop."""
if "event_loop" not in st.session_state:
st.session_state.event_loop = asyncio.new_event_loop()
threading.Thread(target=st.session_state.event_loop.run_forever, daemon=True).start()
return st.session_state.event_loop
def display_chat_messages(messages,user_img,assistant_img):
"""Displays chat messages in the Streamlit app."""
chat_container = st.container()
with chat_container:
for message in messages:
avatar = user_img if message["role"] == "user" else assistant_img
with st.chat_message(message["role"], avatar=avatar):
st.markdown(message["content"])
def poll_for_updates(client: AsyncPinionAIClient, timeout: int, http_poll_start: int = 30, http_poll_interval: int = 5):
"""Polls for updates and returns True if a rerun is needed."""
start_time = time.time()
next_http_poll_time = start_time + http_poll_start
while time.time() - start_time < timeout:
# Primary check: Has a gRPC message arrived recently?
if (time.time() - client._grpc_last_update_time) < 2.0:
return True
# Fallback check: Poll HTTP endpoint no response in a while.
now = time.time()
if now >= next_http_poll_time:
try:
lastmodified_server, _ = run_coroutine_in_event_loop(client.get_latest_session_modification_time())
if lastmodified_server and lastmodified_server != client.last_session_post_modified:
return True
# Schedule the next poll
next_http_poll_time = now + http_poll_interval
except Exception as e:
# Using print instead of st.warning to avoid cluttering the UI
st.warning(f"Warning: Could not check for session updates: {e}")
# Don't hammer on failure, schedule next poll
next_http_poll_time = now + http_poll_interval
time.sleep(0.1) # Prevent busy-waiting
return False # Timeout
def ensure_grpc_is_active(client: AsyncPinionAIClient):
"""
For Live Agent Discussion. Checks if the gRPC client is active
and starts it if not. Makes the app fork-safe.
"""
if not client._grpc_stub:
try:
is_started = run_coroutine_in_event_loop(client.start_grpc_client_listener(sender_id="user"))
if is_started:
st.info("Connecting to live agent...")
return True
else:
st.error("Could not connect to live agent service.")
return False
except Exception as e:
st.error(f"Failed to start gRPC listener: {e}")
return False
return True # Already active
# --- Initialize PinionAIClient ---
st.set_page_config(
page_title="PinionAI Chat",
page_icon="assets/favicon.ico",
menu_items={
'Get help': 'https://docs.pinionai.com/',
'Report a bug': 'https://www.pinionai.com/contact',
'About': 'Use **[PinionAI](https://www.pinionai.com)** as your low-code, opinionated AI Agent Platform. Delivering controlled AI Agents that work seamlessly with existing business infrastructure, and targeting topics you desire, PinionAI performs actions, delivers information using all major models, and offers privacy and security built in. \n\n**PinionAI LLC**, All rights reserved. Version: `0.3.1`'
},
layout="wide"
)
# state for handling private AIA files
if 'awaiting_key_secret' not in st.session_state:
st.session_state.awaiting_key_secret = False
if 'merging_aia' not in st.session_state:
st.session_state.merging_aia = False
if 'uploaded_file_bytes' not in st.session_state:
st.session_state.uploaded_file_bytes = None
if not os.environ.get("agent_id"):
if st.session_state.awaiting_key_secret or st.session_state.get("merging_aia"):
st.warning("This AIA file is private and requires a secret key to decrypt.")
with st.form("key_secret_form"):
key_secret = st.text_input("Enter the secret key:", type="password")
col1, col2 = st.columns(2)
with col1:
button_label = "Unlock and Merge Agent" if st.session_state.get("merging_aia") else "Unlock and Load Agent"
if st.form_submit_button(button_label, width="stretch"):
if not key_secret:
st.error("Please enter a secret key.")
else:
if st.session_state.get("merging_aia"):
with st.spinner("Decrypting and merging agent..."):
try:
file_bytes = st.session_state.uploaded_file_bytes
stringio = StringIO(file_bytes.decode("utf-8"))
result_msg = run_coroutine_in_event_loop(st.session_state.pinion_client.add_agent_from_aia(
file_stream=stringio.read(),
key_secret=key_secret
))
if "Error" not in result_msg:
st.session_state.merging_aia = False
st.session_state.uploaded_file_bytes = None
st.success(result_msg)
time.sleep(1)
st.rerun()
else:
st.error(result_msg)
except Exception as e:
st.error(f"Merge failed: {e}")
else:
# Clear main session state before loading new agent
keys_to_keep = ["logged_in", "user_login", "user_email", "accountSelectedUId", "accountPermissions", "awaiting_key_secret", "uploaded_file_bytes"]
keys_to_delete = [key for key in st.session_state.keys() if key not in keys_to_keep]
for key in keys_to_delete:
del st.session_state[key]
with st.spinner("Decrypting and loading agent..."):
try:
file_bytes = st.session_state.uploaded_file_bytes
stringio = StringIO(file_bytes.decode("utf-8"))
client, init_message = run_coroutine_in_event_loop(AsyncPinionAIClient.create_from_stream(
file_stream=stringio.read(),
host_url=os.environ.get("host_url"),
key_secret=key_secret
))
if client:
st.session_state.pinion_client = client
st.session_state.awaiting_key_secret = False # Clean up temp state on success
st.session_state.uploaded_file_bytes = None
st.success("Agent loaded successfully!")
st.rerun()
else:
st.error(f"Failed to load agent: {init_message}")
st.stop()
except Exception as e:
st.error(f"Failed to load agent with provided key: {e}")
st.stop()
with col2:
if st.form_submit_button("Cancel", width="stretch"):
# Clean up temp state and go back
st.session_state.awaiting_key_secret = False
st.session_state.merging_aia = False
st.session_state.uploaded_file_bytes = None
st.rerun()
st.stop()
else:
with st.form(f"agent_file"):
uploaded_file = st.file_uploader("Upload AIA agent file or shortcut", type="aia", accept_multiple_files=False,
help="Select the File version of the agent to test.")
col_b1, col_b2 = st.columns(2)
with col_b1:
if st.form_submit_button("Load AIA Agent", help="Reset the test agent and clear the chat history."):
if uploaded_file is None:
st.error("Please upload an agent file first.")
st.stop()
else: # Session state is cleared before attempting to load
# keys_to_keep = ["logged_in", "user_login", "user_email", "accountSelectedUId", "accountPermissions"]
keys_to_delete = [key for key in st.session_state.keys()] # if key not in keys_to_keep]
for key in keys_to_delete:
del st.session_state[key]
try:
file_bytes = uploaded_file.getvalue()
stringio = StringIO(file_bytes.decode("utf-8"))
client, init_message = run_coroutine_in_event_loop(AsyncPinionAIClient.create_from_stream(
file_stream=stringio.read(),
host_url=os.environ.get("host_url")
))
if init_message == 'key_secret required for private version':
st.session_state.awaiting_key_secret = True
st.session_state.uploaded_file_bytes = file_bytes
st.rerun()
elif client:
st.session_state.pinion_client = client
st.rerun() # Rerun to load the new agent's chat
else:
st.error(f"Failed to load agent: {init_message}")
st.stop()
except Exception as e:
st.error(f"Failed to initialize PinionAI client from AIA file: {e}")
st.stop()
elif st.session_state.get("pinion_client") is None:
st.info("Please upload an AIA file or shortcut.")
st.stop()
with col_b2:
if st.form_submit_button("Add AIA Agent", help="Add this AIA file's capabilities to your current session."):
if uploaded_file is None:
st.error("Please upload an agent file first.")
elif "pinion_client" not in st.session_state:
st.error("No active session to merge into.")
else:
try:
file_bytes = uploaded_file.getvalue()
stringio = StringIO(file_bytes.decode("utf-8"))
result_msg = run_coroutine_in_event_loop(st.session_state.pinion_client.add_agent_from_aia(
file_stream=stringio.read()
))
if result_msg == 'key_secret required for private version':
# Set flag and store file bytes to show key form
st.session_state.merging_aia = True
st.session_state.uploaded_file_bytes = file_bytes
st.rerun()
elif "Error" not in result_msg:
st.success(result_msg)
time.sleep(1)
st.rerun()
else:
st.error(result_msg)
except Exception as e:
st.error(f"Merge analysis failed: {e}")
else:
if "pinion_client" not in st.session_state:
try:
st.session_state.pinion_client = run_coroutine_in_event_loop(AsyncPinionAIClient.create(
agent_id=os.environ.get("agent_id"),
host_url=os.environ.get("host_url"),
client_id=os.environ.get("client_id"),
client_secret=os.environ.get("client_secret"),
version=os.environ.get("version", None) # Change to serve specific version (draft, development, test, live, archived). None loads latest in progress.
))
if not st.session_state.pinion_client.chat_messages and st.session_state.pinion_client.var.get("agentStart"):
st.session_state.pinion_client.add_message_to_history(
"assistant", st.session_state.pinion_client.var["agentStart"]
)
except PinionAIConfigurationError as e:
st.error(f"Failed to initialize PinionAI client: {e}")
st.stop()
if st.session_state.pinion_client:
client: AsyncPinionAIClient = st.session_state.pinion_client
var = client.var # Convenience to the client's var dictionary
else:
st.stop()
if "end_chat_clicked" not in st.session_state:
st.session_state.end_chat_clicked = False
try:
assistant_img = var["assistImage"]
user_img = var["userImage"]
except KeyError as e:
st.error(f"Error loading image URLs from agent configuration: Missing key {e}. Agent configuration might be incomplete.")
st.stop()
if st.session_state.end_chat_clicked:
st.write("Your conversation has ended.")
st.stop()
# --- UI Layout ---
col1, col2 = st.columns([8, 1])
with col1:
st.header(var["agentTitle"], divider=var["accentColor"])
with col2:
st.image(assistant_img)
st.write(var["agentSubtitle"])
if var["transferAllowed"]:
with st.form(f"chat_status_form_{client.session_id or 'nosession'}"):
col1, col2 = st.columns(2)
with col1:
if st.form_submit_button("Continue"):
st.rerun()
with col2:
if st.form_submit_button("End Chat"):
st.session_state.end_chat_clicked = "yes"
run_coroutine_in_event_loop(client.end_grpc_chat_session())
st.rerun()
# Start gRPC client listener if transfer is requested and not already started
if client.transfer_requested:
ensure_grpc_is_active(client)
display_chat_messages(client.get_chat_messages_for_display(), user_img, assistant_img)
# Accept user input
input_text = None
if var.get("sttAudio"):
if prompt := st.chat_input(var["agentStart"], accept_audio=True):
if prompt.text:
input_text = prompt.text
if prompt.audio:
with st.chat_message("assistant", avatar=assistant_img):
with st.spinner("Processing audio..."):
# Convert audio to text
input_text = run_coroutine_in_event_loop(client.convert_audio_to_text(prompt.audio))
else:
if prompt_str := st.chat_input(var["agentStart"]):
input_text = prompt_str
if input_text is not None:
client.add_message_to_history("user", input_text)
with st.chat_message("user", avatar=user_img):
st.markdown(input_text)
if client.transfer_requested: # LIVE AGENT MODE
if ensure_grpc_is_active(client):
run_coroutine_in_event_loop(client.update_pinion_session())
run_coroutine_in_event_loop(client.send_grpc_message(input_text))
# Poll for a response from the agent before rerunning
if poll_for_updates(client, timeout=180):
st.rerun()
else:
st.warning("No new messages in the last 3 minutes. Please click Continue or End Chat.")
else: # AI AGENT MODE
with st.chat_message("assistant", avatar=assistant_img):
with st.spinner("Thinking..."):
full_ai_response_string = run_coroutine_in_event_loop(client.process_user_input(input_text, sender="user"))
st.markdown(full_ai_response_string)
# The client's process_user_input method already adds the assistant's response to its chat_messages
run_coroutine_in_event_loop(client.update_pinion_session())
if var.get("ttsAudio"):
with st.spinner("Generating audio..."):
try:
audio_bytes = run_coroutine_in_event_loop(client.convert_text_to_audio(full_ai_response_string))
if audio_bytes:
# Auto-detect format from magic bytes (RIFF = WAV, OggS = OGG, default to MP3)
audio_format = "audio/mp3"
if audio_bytes.startswith(b"RIFF"):
audio_format = "audio/wav"
elif audio_bytes.startswith(b"OggS"):
audio_format = "audio/ogg"
st.audio(audio_bytes, format=audio_format, autoplay=True)
except Exception as e:
st.error(f"Failed to generate TTS audio: {e}")
# Handle if a next_intent was set by the AI's processing. Next_intent turn handled internally
if client.next_intent:
with st.chat_message("assistant", avatar=assistant_img):
with st.spinner("Thinking..."):
# Process the next_intent (user_input might be empty or the next_intent itself)
full_next_intent_response_string = run_coroutine_in_event_loop(client.process_user_input(user_input="", sender="user"))
st.markdown(full_next_intent_response_string)
run_coroutine_in_event_loop(client.update_pinion_session())
if var.get("ttsAudio"):
with st.spinner("Generating audio..."):
try:
audio_bytes = run_coroutine_in_event_loop(client.convert_text_to_audio(full_next_intent_response_string))
if audio_bytes:
# Auto-detect format from magic bytes (RIFF = WAV, OggS = OGG, default to MP3)
audio_format = "audio/mp3"
if audio_bytes.startswith(b"RIFF"):
audio_format = "audio/wav"
elif audio_bytes.startswith(b"OggS"):
audio_format = "audio/ogg"
st.audio(audio_bytes, format=audio_format, autoplay=True)
except Exception as e:
st.error(f"Failed to generate TTS audio: {e}")
if client.transfer_requested:
# Start gRPC client listener if agent transfer is requested
if ensure_grpc_is_active(client):
st.info("Transfer to live agent initiated... Waiting for agent to connect.")
# Poll for the first message from the agent
if poll_for_updates(client, timeout=180):
st.rerun()
else:
st.warning("No new messages in the last 3 minutes. Please click Continue or End Chat.")
else:
st.error("Could not connect to live agent service for transfer.")
elif client.transfer_requested: # If transfer was already active, and AI responded (e.g. fallback)
if poll_for_updates(client, timeout=180):
st.rerun()
else:
st.warning("No new messages in the last 3 minutes. Please click Continue or End Chat.")