Using the Python SDK

The Python SDK provides an asyncio-native client for managing Sec-Gemini sessions programmatically. All methods are async – wrap with asyncio.run() in synchronous scripts, or use await directly in notebooks.

Send Your First Prompt via the API

Create a session, send a prompt, and stream the agent’s work:

🐍 Send Your First Prompt via the API

import asyncio
from sec_gemini import SecGemini

async def main():
  async with SecGemini(api_key="YOUR_API_KEY") as client:
      session = await client.sessions.create()
      await session.prompt("Analyze the email security posture of example.com")

      async for msg in session.messages.stream():
          msg_type = msg.get("message_type", "")
          content = msg.get("content", "")

          if msg_type == "MESSAGE_TYPE_RESPONSE":
              print(f"Agent: {content}")
          elif msg_type == "MESSAGE_TYPE_THOUGHT":
              print(f"  (thinking: {content[:100]})")
          elif msg_type == "MESSAGE_TYPE_TOOL_CALL":
              print(f"  [tool] {msg.get('title', '')}")

asyncio.run(main())

Terminal Output

Check and Resume Sessions

Sessions persist server-side. If your client disconnects, the agent keeps working. Reconnect to check status or resume streaming:

🐍 Check and Resume Sessions

import asyncio
from sec_gemini import SecGemini

async def main():
  async with SecGemini(api_key="YOUR_API_KEY") as client:
      # List all sessions
      sessions = await client.sessions.list()
      print("Active Sessions:")
      for s in sessions:
          print(f"  {s.id[:12]}  {s.name:<30}  {s.status}")

      # Resume streaming from an existing session
      session = await client.sessions.get("sess_demo_12345")
      print("\nResuming Message Stream:")
      async for msg in session.messages.stream():
          if msg.get("content"):
              print(f"  > {msg.get('content')}")

asyncio.run(main())

Terminal Output

You can also pause, resume, or cancel:

await session.pause()
await session.resume()
await session.cancel()

Attach Files

Upload files for the agent to reference during a session:

await session.files.upload("/path/to/report.pdf")

files = await session.files.list()
for f in files:
    print(f.filename)

await session.files.delete("report.pdf")

The agent works best with text-based formats (TXT, JSON, CSV, YAML, XML, logs, source code). PDFs are supported with automatic text extraction. Files expire after 7 days.

Add Your Own Tools

Use Bring Your Own Tools (BYOT) to connect local tools or custom FastMCP servers on your machine to your SDK sessions.

Add Skills

Skills are markdown instructions that teach the agent specific workflows:

await client.skills.upload(
    name="vuln-triage.md",
    content="""---
name: vuln-triage
description: Triage and prioritize CVEs
---
When given CVEs, prioritize by exploitability,
check for patches, and present a ranked table.
"""
)

Handle Errors

All exceptions inherit from SdkError:

from sec_gemini import SecGemini, SdkError, ConnectionLostError

try:
    async with SecGemini(api_key="YOUR_API_KEY") as client:
        session = await client.sessions.create()
        await session.prompt("Analyze example.com")
        async for msg in session.messages.stream():
            print(msg.get("content", ""))
except ConnectionLostError:
    print("Connection lost -- reconnect and stream again to resume.")
except SdkError as e:
    print(f"SDK error: {e}")

See the SDK API Reference for the complete API surface, message types, and exception hierarchy.