writing-streamlit-apps

为 PostHog 沙盒编写 Streamlit 应用

你编写的源代码会成为沙盒 Streamlit 1.31 运行时根目录下的 app.py。 部署机制（创建／启动／分享）请参阅 managing-streamlit-apps 技能；本技能介绍如何编写代码。

读取 PostHog 数据：posthog_apps.query()

唯一的数据入口是沙盒内的桥接接口：

import posthog_apps

df = posthog_apps.query("SELECT event, count() FROM events GROUP BY event LIMIT 10")

• 接受 HogQL 字符串，返回 pandas DataFrame。
• 失败时会抛出 RuntimeError。错误消息有意保持通用（“Query execution failed”）——桥接接口不会向沙盒返回查询内部信息，因此你无法在应用内诊断错误查询。请捕获异常并通过 st.error(str(e)) 显示，这样查看者看到的是一条消息，而不是堆栈跟踪；请在 SQL 编辑器中测试查询，那里可以看到实际错误。
• 沙盒中不存在 import posthog——模块名是 posthog_apps，有意与 posthog-python SDK 的名称区分开来。
• 桥接接口已预先通过身份验证，可访问应用所属项目；用户代码不会接触令牌，也无需进行任何配置。
• 查询受服务端限制（执行时间 30 s、内存 256 MB），且不会随沙盒规格变化而调整——因此请限制时间范围、使用 LIMIT 限制结果，并在 HogQL 中进行聚合，而不是将原始事件载入 pandas。

针对 Streamlit 的重新运行模型进行设计

每次与控件交互时，Streamlit 都会从头到尾重新运行整个脚本。 这会带来两个后果：

1. 缓存每次桥接调用——如果不缓存，拖动一次滑块就会重新触发每个查询：

   @st.cache_data(ttl=300, show_spinner="Running query...")
   def run_query(hogql: str) -> pd.DataFrame:
       return posthog_apps.query(hogql)
   

2. 使用 st.session_state 保存需要在重新运行后保留的内容——累积的选择、分页游标、“上次刷新”时间戳。模块级变量会在每次交互时重置。

控件可以自然地驱动参数，但绝不要将自由文本控件的值插入 HogQL。 桥接接口会使用版本作者的数据访问权限运行查询，而任何能查看应用的人都可以操作这些控件——把原始 st.text_input 值拼接到查询中，就等于把作者的访问权限交给查看者，让他们编写自己的查询。 请改为限制输入：从你控制的固定列表中选择（例如通过 st.selectbox 提供已知值），或将输入转换为不可能携带 SQL 的类型（例如 int(days)、来自 st.date_input 的 date），并在输入到达查询之前进行验证。

布局和图表

• 首先调用 st.set_page_config(page_title=..., layout="wide")——默认的窄布局会让数据应用浪费大部分屏幕空间。
• 使用 st.columns 并排展示指标，使用 st.tabs 提供不同视图，使用 st.expander 展示详情部分；使用 st.metric 显示关键数字。
• 图表：使用 plotly express 配合 st.plotly_chart(fig, use_container_width=True) 是可靠的默认选择；使用 st.dataframe(df, use_container_width=True) 展示表格。也可以通过 st.pyplot 使用 matplotlib／seaborn。

已安装的内容

镜像内置 Python 3.11，包含：streamlit 1.31、pandas、numpy、polars、plotly、matplotlib、seaborn、scipy、scikit-learn、pyarrow、duckdb、requests、beautifulsoup4、lxml、sqlalchemy、aiohttp。 无法添加依赖：沙盒不会运行 pip（这是有意采取的安全措施，启动时不会执行任意软件包代码），上传的 zip 中即使包含 requirements.txt 也会被接受但随后丢弃。只能导入上面列出的软件包。

结构和运行时限制

• app.py 是入口点。 通过 MCP 的 set-source 流程，你的源代码就是 app.py。辅助模块和数据文件可以通过 files（文本）和 assets（base64）映射随其一起提供；import utils 可以正常工作，因为 Streamlit 会将应用目录加入 sys.path，但进程并不会在该目录中运行，因此应通过 Path(__file__).parent / "data/events.parquet" 打开数据文件，切勿使用裸相对路径。请保持随应用打包的数据较小：每次调用 set-source 时，数据都会以内联 JSON 的形式发送。
• 沙箱是临时的：停止或重启后，写入磁盘的内容都会消失。不要依赖文件保存状态；应根据查询重新计算（并使用缓存），或将状态保存在 st.session_state 中。
• 你的代码以非特权用户身份运行；没有 posthog SDK，也无法向应用传入自定义环境变量或密钥，并且不能指望能够访问通用的外部网络。不要读取 os.environ——其中的内容属于沙箱运行时，而非你的应用。请围绕 posthog_apps.query() 作为数据源来设计。

一个结构规范的最小应用

import pandas as pd
import plotly.express as px
import posthog_apps
import streamlit as st

st.set_page_config(page_title="Events overview", layout="wide")
st.title("Events overview")


@st.cache_data(ttl=300, show_spinner="Running query...")
def run_query(hogql: str) -> pd.DataFrame:
    return posthog_apps.query(hogql)


# A slider is bounded and coerced to int, so it is safe to interpolate.
days = int(st.slider("Days to show", 1, 30, 7))
try:
    daily = run_query(
        f"""
        SELECT toDate(timestamp) AS day, count() AS events
        FROM events
        WHERE timestamp >= now() - INTERVAL {days} DAY
        GROUP BY day ORDER BY day
        """
    )
    st.plotly_chart(px.bar(daily, x="day", y="events"), use_container_width=True)
except RuntimeError as e:
    st.error(str(e))