Python SDK

git clone https://github.com/hasanraiyan/agent-marketplace.git
cd agent-marketplace/sdk-python
python -m venv .venv
.venv/Scripts/pip install .   # .venv/bin/pip on macOS/Linux

pip install -e /path/to/agent-marketplace/sdk-python

pip install personaai

import os
from personaai import PersonaClient

# Control-plane client — acts as the Project itself.
persona = PersonaClient(os.environ["PERSONA_BASE_URL"], credential=os.environ["PERSONA_CREDENTIAL"])

# Runtime-plane client — acts as one specific end user of yours.
user_persona = PersonaClient(
    os.environ["PERSONA_BASE_URL"],
    credential=os.environ["PERSONA_CREDENTIAL"],
    external_user_id=current_user.id,
)

from personaai import PersonaClient

persona = PersonaClient(
    "https://api.personaai.com",
    credential="<keyId>.<secret>",  # minted via Studio
)

who = persona.whoami()
print(who["principalType"], who["domain"])

agent = persona.agents.create(
    {
        "name": "Career Launchpad",
        "systemPrompt": "You help students find internships.",
        "providerId": "...",  # an existing Provider's id
        "visibility": "unlisted",
    }
)

import asyncio
from personaai import AsyncPersonaClient

async def main():
    async with AsyncPersonaClient("https://api.personaai.com", credential="<keyId>.<secret>") as persona:
        who = await persona.whoami()
        print(who["principalType"], who["domain"])

asyncio.run(main())

user_persona = PersonaClient(
    "https://api.personaai.com",
    credential="<keyId>.<secret>",
    external_user_id=current_user.id,
)

thread = user_persona.threads.create({"agentId": agent["_id"]})

who = persona.whoami()

from personaai import PersonaApiError, PersonaAuthError, PersonaValidationError

try:
    persona.agents.create({"name": "", "systemPrompt": "...", "providerId": "..."})
except PersonaValidationError as err:
    print(f"Bad request ({err.code}): {err}")
except PersonaAuthError as err:
    print(f"Auth problem ({err.status_code}): {err}")
except PersonaApiError as err:
    print(f"API error ({err.status_code}/{err.code}): {err}")

patient_client = PersonaClient(
    "...", credential="...", max_retries=5  # retry up to 5 times on 429 before raising
)

provider = persona.providers.create(
    {
        "label": "OpenAI (production)",
        "baseURL": "https://api.openai.com/v1",
        "apiKey": os.environ["OPENAI_API_KEY"],  # encrypted at rest, never returned in any response
        "defaultModel": "gpt-4o-mini",
        "isDefault": True,
    }
)

result = persona.providers.test_connection(provider["id"])
models = persona.providers.get_models(provider["id"])

skill = persona.skills.create(
    {
        "name": "Resume Reviewer",
        "description": "Reviews a resume and suggests concrete edits.",
        "instructions": "You are an expert resume reviewer...",
        "isPublic": False,
        "files": [{"path": "rubric.md", "content": "# Scoring rubric\n..."}],
    }
)

mine = user_persona.skills.list({"scope": "mine", "search": "resume"})
full = persona.skills.get(skill["_id"])
persona.skills.update(skill["_id"], {"description": "Updated description."})
persona.skills.delete(skill["_id"])

agent = persona.agents.create(
    {
        "name": "Career Launchpad",
        "systemPrompt": "You help students find internships.",
        "providerId": provider["id"],
        "description": "Your internship search co-pilot.",
        "category": "productivity",
        "visibility": "unlisted",
        "webSearchEnabled": True,
        "skills": [skill["_id"]],
    }
)

discovered = persona.agents.list({"category": "productivity", "search": "career"})
full = persona.agents.get(agent["_id"])  # skills/mcps/knowledgeBases populated as objects
persona.agents.update(agent["_id"], {"tagline": "Find your next internship."})
persona.agents.delete(agent["_id"])

kb = persona.knowledge.create(
    {
        "name": "Placement Office FAQ",
        "providerId": provider["id"],  # required — no "my default provider" concept on this API
        "chunkSize": 1000,
        "chunkOverlap": 200,
        "topK": 5,
    }
)

with open("faq.pdf", "rb") as f:
    result = persona.knowledge.upload_documents(
        kb["_id"], [{"filename": "faq.pdf", "content": f.read(), "contentType": "application/pdf"}]
    )
# result: {"documentCount": ..., "chunkCount": ..., "files": [{"fileName", "fileSize", "mimeType", "chunkCount"}]}

docs = persona.knowledge.list_documents(kb["_id"])
persona.knowledge.delete_document(kb["_id"], "faq.pdf")

hits = persona.knowledge.search(kb["_id"], "What's the deadline to apply?", top_k=3)
# hits: [{"text", "source", "score"}, ...]

mcp = persona.mcps.create(
    {
        "name": "Internal Ticketing",
        "transport": "http",
        "url": "https://mcp.example.com",
        "authType": "apiKey",
        "apiKey": os.environ["TICKETING_MCP_KEY"],
    }
)

connection = persona.mcps.test_connection(mcp["_id"])
result = persona.mcps.call_tool(mcp["_id"], "create_ticket", {"title": "Bug report"})
resource = persona.mcps.read_resource(mcp["_id"], connection["resources"][0]["uri"])

thread = user_persona.threads.create({"agentId": agent["_id"]})
mine = user_persona.threads.list({"page": 1, "limit": 20})
user_persona.threads.update_title(thread["_id"], "Summer internship search")
messages = user_persona.threads.get_messages(thread["_id"])
user_persona.threads.delete(thread["_id"])

with open("resume.pdf", "rb") as f:
    file = user_persona.files.upload(
        {
            "filename": "resume.pdf",
            "content": f.read(),
            "contentType": "application/pdf",
            "threadId": thread["_id"],  # optional — associates the file with a conversation
        }
    )

mine = user_persona.files.list({"page": 1, "limit": 20})

response = user_persona.files.download(file["id"])  # raw httpx.Response, not JSON
bytes_ = response.content

user_persona.files.delete(file["id"])

from personaai import EventType

# Full event stream, for building your own UI.
for event in user_persona.chat.stream(
    agent["_id"], [{"role": "user", "content": "What internships are open right now?"}]
):
    if event["type"] == EventType.TEXT_MESSAGE_CHUNK and event.get("delta"):
        print(event["delta"], end="")

# Or the convenience wrapper — drains the stream, returns the final text.
result = user_persona.chat.send_message(
    agent["_id"], [{"role": "user", "content": "What internships are open right now?"}]
)
print(result["text"])

async for event in user_persona.chat.stream(agent["_id"], messages):
    ...

result = await user_persona.chat.send_message(agent["_id"], messages)

for event in user_persona.chat.stream(agent["_id"], messages):
    if event["type"] == EventType.TEXT_MESSAGE_CHUNK:
        print(event.get("delta", ""), end="")
    elif event["type"] == EventType.TOOL_CALL_RESULT:
        print("tool result:", event)
    elif event["type"] == EventType.RUN_FINISHED:
        print("done")

result = user_persona.chat.send_message(agent["_id"], messages)

if result["interrupt"]:
    print(result["interrupt"]["kind"])   # "hitl" | "clarification"
    print(result["interrupt"]["value"])  # whatever detail the interrupt carries

if result["interrupt"]:
    user_persona.chat.send_message(
        agent["_id"],
        [],
        thread_id=thread["_id"],
        resume={"decisions": [{"action": "delete_agent", "decision": "approve"}]},
    )

# persona.py
import os
from personaai import PersonaClient

persona = PersonaClient(os.environ["PERSONA_BASE_URL"], credential=os.environ["PERSONA_CREDENTIAL"])

# app.py
import os
from flask import Flask, request, jsonify
from personaai import PersonaClient

app = Flask(__name__)

@app.post("/api/chat")
def chat():
    user_persona = PersonaClient(
        os.environ["PERSONA_BASE_URL"],
        credential=os.environ["PERSONA_CREDENTIAL"],
        external_user_id=request.json["userId"],
    )
    result = user_persona.chat.send_message(request.json["agentId"], request.json["messages"])
    return jsonify(result)

# deps.py
import os
from personaai import AsyncPersonaClient

def get_persona(external_user_id: str | None = None) -> AsyncPersonaClient:
    return AsyncPersonaClient(
        os.environ["PERSONA_BASE_URL"],
        credential=os.environ["PERSONA_CREDENTIAL"],
        external_user_id=external_user_id,
    )

# main.py
from fastapi import Depends, FastAPI
from deps import get_persona

app = FastAPI()

@app.post("/api/chat")
async def chat(body: dict, current_user_id: str = Depends(get_current_user_id)):
    async with get_persona(external_user_id=current_user_id) as persona:
        return await persona.chat.send_message(body["agentId"], body["messages"])

# yourapp/persona.py
from django.conf import settings
from personaai import PersonaClient

persona = PersonaClient(settings.PERSONA_BASE_URL, credential=settings.PERSONA_CREDENTIAL)