pycelest — AI Agent Framework

Build · Orchestrate · Defend · Connect A production-ready Python framework for safe, capable, and observable AI agents.

What is pycelest?

pycelest is a Python framework for building AI agents that are safe, capable, and observable — designed from day one for production deployment, not just demos.

Unlike frameworks that bolt guardrails on as an afterthought, pycelest puts security, traceability, and reliability at the center of its architecture.

Features

Installation

# Core framework
pip install pycelest

# With providers
pip install pycelest[anthropic]   # Claude
pip install pycelest[openai]      # GPT-4o, Gemini, DeepSeek, Mistral, Grok

# Extras
pip install pycelest[mcp]         # MCP client + server
pip install pycelest[langfuse]    # Observability
pip install pycelest[gmail]       # Gmail connector
pip install pycelest[rag]         # ChromaDB RAG

# Everything
pip install pycelest[all]

Quick Start

import asyncio
from celest import SessionManager, SessionConfig
from celest.providers import AnthropicAdapter

config = SessionConfig(
    system_prompt="You are a helpful, careful AI agent.",
    max_iterations=8,
)

session = SessionManager(
    config=config,
    provider=AnthropicAdapter(model="claude-sonnet-4-6", api_key="sk-ant-..."),
)

result = asyncio.run(session.run("Plan a 3-day trip to Kyoto"))
print(result.response)

Local Models (no API key)

from celest.providers import OllamaAdapter

session = SessionManager(
    config=config,
    provider=OllamaAdapter(model="llama3.2"),
)

Install Ollama from ollama.com, then ollama pull llama3.2.

With Tools

from celest.tools import FunctionTool

@FunctionTool.register(description="Search the web for current information")
async def web_search(query: str) -> str:
    ...

session = SessionManager(config=config, provider=provider, tools=[web_search])
result = await session.run("What are the latest AI news?")

MCP Client — consume any MCP server

from celest.tools.mcp import MCPAdapter

# Stdio (local subprocess)
async with MCPAdapter(command="uvx", args=["mcp-server-time"]) as mcp:
    tools = await mcp.discover_tools()
    session = SessionManager(config=config, provider=provider, tools=tools)

# SSE (remote server) with auth + auto-reconnect
async with MCPAdapter(
    url="https://my-mcp-server.com/sse",
    headers={"Authorization": "Bearer sk-..."},
    auto_reconnect=True,
) as mcp:
    tools = await mcp.discover_tools()

MCP Server — expose your agent as an MCP server

from celest import SessionManager, SessionConfig, MCPServer
from celest.providers import AnthropicAdapter

def make_session():
    return SessionManager(
        config=SessionConfig(system_prompt="You are a helpful assistant."),
        provider=AnthropicAdapter(model="claude-sonnet-4-6", api_key="sk-ant-..."),
    )

server = MCPServer(session_factory=make_session, name="my-agent")
server.run_stdio()   # or server.run_sse(port=8080)

Add to claude_desktop_config.json:

{
  "mcpServers": {
    "my-agent": {
      "command": "python",
      "args": ["my_agent.py"]
    }
  }
}

Native Connectors

from celest.connectors.native.gmail import GmailConnector
from celest.connectors.native.slack import SlackConnector
from celest.connectors.native.notion import NotionConnector

gmail = GmailConnector(access_token="ya29....")
tools = gmail.get_tools()   # gmail_send, gmail_read

slack = SlackConnector(bot_token="xoxb-...")
tools += slack.get_tools()  # slack_send, slack_read

session = SessionManager(config=config, provider=provider, tools=tools)

Observability with Langfuse

pip install pycelest[langfuse]

from celest.plugins.langfuse import LangfusePlugin

plugin = LangfusePlugin(
    public_key="pk-lf-...",
    secret_key="sk-lf-...",
    host="https://cloud.langfuse.com",
    agent_name="my-agent",
)

session = SessionManager(config=config, provider=provider, plugins=[plugin])
# Every run is now traced in Langfuse: turns, tool calls, tokens, latency

YAML Config

# celest.yaml
system_prompt: "You are a helpful AI agent"
max_iterations: 8
provider: anthropic
model: claude-sonnet-4-6
guardrails:
  tool_firewall: ask   # accept | deny | ask

session = SessionManager.from_yaml("celest.yaml")
result = await session.run("Your task here")

CLI

celest init                           # Generate starter celest.yaml
celest run celest.yaml "Your prompt"  # Run agent from config

Changelog

License

Proprietary © 2026 Lovanirainy Theogene Eddy Celestin RAFANOMEZANA — flycelest.com