import os
import cpsl
app = cpsl.App(
name="integrations-demo",
image=cpsl.Image(),
secrets=["OPENAI_API_KEY"],
filesystems={"/data": cpsl.FileSystem("reports")},
)
app.add_integration(
"github",
client_id=cpsl.Secret.from_name("GITHUB_CLIENT_ID"),
client_secret=cpsl.Secret.from_name("GITHUB_CLIENT_SECRET"),
scopes=["repo", "read:user"],
)
app.add_integration("aws")
@app.boot()
async def on_boot():
os.makedirs("/data/reports", exist_ok=True)
@app.message()
async def handle(session: cpsl.Session, msg: cpsl.Message):
text = (msg.text or "").strip().lower()
if text == "connect github":
cred = await session.prompt_integration(
"github",
reason="Need GitHub access to inspect repositories",
)
await session.reply(f"Connected github with scopes: {cred.scopes}")
return
if text == "connect aws":
cred = await session.prompt_integration(
"aws",
reason="Need AWS credentials to inspect infrastructure",
)
await session.reply(f"Connected aws for region {cred.fields.get('region')}")
return
if text == "hint github":
await session.show_integration(
"github",
reason="Connect GitHub to unlock repository analysis",
)
await session.reply("Rendered a non-blocking integration card.")
return
if text == "upload csv":
upload = await session.prompt_file(
message="Upload a CSV file",
accept=".csv,text/csv",
path="/data/uploads",
)
await session.reply(f"Uploaded {upload.name} to {upload.path}")
return
if text == "write report":
api_key = cpsl.Secret.from_name("OPENAI_API_KEY").value
with open("/data/reports/summary.txt", "w") as f:
f.write(f"Secret loaded: {bool(api_key)}\n")
await session.reply("Wrote /data/reports/summary.txt")
return
await session.reply(
"Try: connect github, connect aws, hint github, upload csv, or write report"
)