agentica-sdk by parcadei/continuous-claude-v3
npx skills add https://github.com/parcadei/continuous-claude-v3 --skill agentica-sdk使用 Agentica 框架在 Python 中构建 AI 智能体。智能体可以实现函数、维护状态、使用工具并相互协调。
在以下情况下使用此技能:
from agentica import agentic
@agentic()
async def add(a: int, b: int) -> int:
"""返回 a 和 b 的和"""
...
result = await add(1, 2) # 智能体计算: 3
from agentica import spawn
agent = await spawn(premise="你是一个只说真话的人。")
result: bool = await agent.call(bool, "地球是平的")
# 返回: False
# 字符串(默认)
result = await agent.call("2+2 等于多少?")
# 类型化输出
result: int = await agent.call(int, "2+2 等于多少?")
result: dict[str, int] = await agent.call(dict[str, int], "统计项目数量")
# 仅副作用
await agent.call(None, "给 John 发送消息")
广告位招租
在这里展示您的产品或服务
触达数万 AI 开发者,精准高效
# Premise: 添加到默认系统提示
agent = await spawn(premise="你是一位数学专家。")
# System: 完全控制(替换默认提示)
agent = await spawn(system="你是一个只响应 JSON 的应答者。")
from agentica import agentic, spawn
# 在装饰器中
@agentic(scope={'web_search': web_search_fn})
async def researcher(query: str) -> str:
"""研究一个主题。"""
...
# 在 spawn 中
agent = await spawn(
premise="数据分析器",
scope={"analyze": custom_analyzer}
)
# 每次调用的作用域
result = await agent.call(
dict[str, int],
"分析数据集",
dataset=data, # 作为 'dataset' 可用
analyzer=custom_fn # 作为 'analyzer' 可用
)
from slack_sdk import WebClient
slack = WebClient(token=SLACK_TOKEN)
# 提取特定方法
@agentic(scope={
'list_users': slack.users_list,
'send_message': slack.chat_postMessage
})
async def team_notifier(message: str) -> None:
"""发送团队通知。"""
...
agent = await spawn(premise="乐于助人的助手")
__init__)from agentica.agent import Agent
class CustomAgent:
def __init__(self):
# 同步 - 使用 Agent() 而不是 spawn()
self._brain = Agent(
premise="专业助手",
scope={"tool": some_tool}
)
async def run(self, task: str) -> str:
return await self._brain(str, task)
# 在 spawn 中
agent = await spawn(
premise="快速响应",
model="openai:gpt-5" # 默认: openai:gpt-4.1
)
# 在装饰器中
@agentic(model="anthropic:claude-sonnet-4.5")
async def analyze(text: str) -> dict:
"""分析文本。"""
...
可用模型:
openai:gpt-3.5-turbo, openai:gpt-4o, openai:gpt-4.1, openai:gpt-5anthropic:claude-sonnet-4, anthropic:claude-opus-4.1anthropic:claude-sonnet-4.5, anthropic:claude-opus-4.5google/gemini-2.5-flash)@agentic(persist=True)
async def chatbot(message: str) -> str:
"""记住对话历史。"""
...
await chatbot("我的名字是 Alice")
await chatbot("我叫什么名字?") # 知道: Alice
对于 spawn() 生成的智能体,同一实例的多次调用之间状态是自动保持的。
from agentica import spawn, MaxTokens
# 简单限制
agent = await spawn(
premise="简短回答",
max_tokens=500
)
# 细粒度控制
agent = await spawn(
premise="受控输出",
max_tokens=MaxTokens(
per_invocation=5000, # 所有轮次的总数
per_round=1000, # 每轮推理
rounds=5 # 最大推理轮次
)
)
from agentica import spawn, last_usage, total_usage
agent = await spawn(premise="你乐于助人。")
await agent.call(str, "你好!")
# 智能体方法
usage = agent.last_usage()
print(f"上次: {usage.input_tokens} 输入, {usage.output_tokens} 输出")
usage = agent.total_usage()
print(f"总计: {usage.total_tokens} 已处理")
# 对于 @agentic 函数
@agentic()
async def my_fn(x: str) -> str: ...
await my_fn("test")
print(last_usage(my_fn))
print(total_usage(my_fn))
from agentica import spawn
from agentica.logging.loggers import StreamLogger
import asyncio
agent = await spawn(premise="你乐于助人。")
stream = StreamLogger()
with stream:
result = asyncio.create_task(
agent.call(bool, "巴黎是法国的首都吗?")
)
# 首先消费流以获取实时输出
async for chunk in stream:
print(chunk.content, end="", flush=True)
# chunk.role 是 'user', 'agent', 或 'system'
# 然后等待结果
final = await result
from agentica import spawn, agentic
# 通过配置文件
agent = await spawn(
premise="使用工具的智能体",
mcp="path/to/mcp_config.json"
)
@agentic(mcp="path/to/mcp_config.json")
async def tool_user(query: str) -> str:
"""使用 MCP 工具。"""
...
mcp_config.json 格式:
{
"mcpServers": {
"tavily-remote-mcp": {
"command": "npx -y mcp-remote https://mcp.tavily.com/mcp/?tavilyApiKey=<key>",
"env": {}
}
}
}
./logs/agent-<id>.logfrom agentica.logging.loggers import FileLogger, PrintLogger
from agentica.logging.agent_logger import NoLogging
# 仅文件
with FileLogger():
agent = await spawn(premise="调试智能体")
await agent.call(int, "计算")
# 静默
with NoLogging():
agent = await spawn(premise="静默智能体")
# 监听器在 agent_listener 子模块中(不从 agentica.logging 导出)
from agentica.logging.agent_listener import (
PrintOnlyListener, # 仅控制台输出
FileOnlyListener, # 仅文件日志
StandardListener, # 控制台 + 文件(默认)
NoopListener, # 静默 - 无日志
)
agent = await spawn(
premise="自定义日志记录",
listener=PrintOnlyListener
)
# 静默智能体
agent = await spawn(
premise="静默智能体",
listener=NoopListener
)
from agentica.logging.agent_listener import (
set_default_agent_listener,
get_default_agent_listener,
PrintOnlyListener,
)
set_default_agent_listener(PrintOnlyListener)
set_default_agent_listener(None) # 禁用所有
from agentica.errors import (
AgenticaError, # 所有 SDK 错误的基类
RateLimitError, # 速率限制
InferenceError, # 来自推理的 HTTP 错误
MaxTokensError, # 超出令牌限制
MaxRoundsError, # 超出最大推理轮次
ContentFilteringError, # 内容被过滤
APIConnectionError, # 网络问题
APITimeoutError, # 请求超时
InsufficientCreditsError,# 积分不足
OverloadedError, # 服务器过载
ServerError, # 通用服务器错误
)
try:
result = await agent.call(str, "做某事")
except RateLimitError:
await asyncio.sleep(60)
result = await agent.call(str, "做某事")
except MaxTokensError:
# 减少作用域或增加限制
pass
except ContentFilteringError:
# 内容被过滤
pass
except InferenceError as e:
logger.error(f"推理失败: {e}")
except AgenticaError as e:
logger.error(f"SDK 错误: {e}")
class DataValidationError(Exception):
"""无效的输入数据。"""
pass
@agentic(DataValidationError) # 传递异常类型
async def analyze(data: str) -> dict:
"""
分析数据。
抛出:
DataValidationError: 如果数据格式错误
"""
...
try:
result = await analyze(raw_data)
except DataValidationError as e:
logger.warning(f"无效: {e}")
from agentica.agent import Agent
class ResearchAgent:
def __init__(self, web_search_fn):
self._brain = Agent(
premise="研究助手。",
scope={"web_search": web_search_fn}
)
async def research(self, topic: str) -> str:
return await self._brain(str, f"研究: {topic}")
async def summarize(self, text: str) -> str:
return await self._brain(str, f"总结: {text}")
class LeadResearcher:
def __init__(self):
self._brain = Agent(
premise="协调子智能体之间的研究。",
scope={"SubAgent": ResearchAgent}
)
async def __call__(self, query: str) -> str:
return await self._brain(str, query)
lead = LeadResearcher()
report = await lead("研究 2025 年 AI 智能体框架")
from agentica import initialize_tracing
# 初始化追踪(返回 TracerProvider)
tracer = initialize_tracing(
service_name="my-agent-app",
environment="development", # 可选
tempo_endpoint="http://localhost:4317", # 可选: Grafana Tempo
organization_id="my-org", # 可选
log_level="INFO", # DEBUG, INFO, WARNING, ERROR
instrument_httpx=False, # 可选: 追踪 HTTP 调用
)
from agentica import enable_sdk_logging
# 启用内部 SDK 日志(用于调试 SDK 本身)
disable_fn = enable_sdk_logging(log_tags="1")
# ... 运行智能体 ...
disable_fn() # 完成后禁用
# 从 agentica 的主要导入
from agentica import (
# 核心
Agent, # 同步智能体类
agentic, # @agentic 装饰器
spawn, # 异步智能体创建
# 配置
ModelStrings, # 模型字符串类型提示
AgenticFunction, # 智能体函数类型
# 令牌跟踪
last_usage, # 获取上次调用的令牌使用情况
total_usage, # 获取累计令牌使用情况
# 追踪/日志记录
initialize_tracing, # OpenTelemetry 设置
enable_sdk_logging, # SDK 调试日志
# 版本
__version__, # "0.3.1"
)
使用 Agentica 前:
@agentic() 的函数必须是 asyncspawn() 返回可等待对象 - 使用 await spawn(...)agent.call() 是可等待的 - 使用 await agent.call(...)call() 的第一个参数是返回类型,第二个是提示字符串@agentic 中使用 persist=True 以获得对话记忆__init__ 中使用 Agent()(而不是 spawn())agentica.logging.agent_listener 导入监听器(而不是 agentica.logging)每周安装
198
仓库
GitHub 星标
3.6K
首次出现
2026年1月22日
安全审计
安装于
opencode191
codex191
gemini-cli189
cursor187
github-copilot185
amp181
Build AI agents in Python using the Agentica framework. Agents can implement functions, maintain state, use tools, and coordinate with each other.
Use this skill when:
from agentica import agentic
@agentic()
async def add(a: int, b: int) -> int:
"""Returns the sum of a and b"""
...
result = await add(1, 2) # Agent computes: 3
from agentica import spawn
agent = await spawn(premise="You are a truth-teller.")
result: bool = await agent.call(bool, "The Earth is flat")
# Returns: False
# String (default)
result = await agent.call("What is 2+2?")
# Typed output
result: int = await agent.call(int, "What is 2+2?")
result: dict[str, int] = await agent.call(dict[str, int], "Count items")
# Side-effects only
await agent.call(None, "Send message to John")
# Premise: adds to default system prompt
agent = await spawn(premise="You are a math expert.")
# System: full control (replaces default)
agent = await spawn(system="You are a JSON-only responder.")
from agentica import agentic, spawn
# In decorator
@agentic(scope={'web_search': web_search_fn})
async def researcher(query: str) -> str:
"""Research a topic."""
...
# In spawn
agent = await spawn(
premise="Data analyzer",
scope={"analyze": custom_analyzer}
)
# Per-call scope
result = await agent.call(
dict[str, int],
"Analyze the dataset",
dataset=data, # Available as 'dataset'
analyzer=custom_fn # Available as 'analyzer'
)
from slack_sdk import WebClient
slack = WebClient(token=SLACK_TOKEN)
# Extract specific methods
@agentic(scope={
'list_users': slack.users_list,
'send_message': slack.chat_postMessage
})
async def team_notifier(message: str) -> None:
"""Send team notifications."""
...
agent = await spawn(premise="Helpful assistant")
__init__)from agentica.agent import Agent
class CustomAgent:
def __init__(self):
# Synchronous - use Agent() not spawn()
self._brain = Agent(
premise="Specialized assistant",
scope={"tool": some_tool}
)
async def run(self, task: str) -> str:
return await self._brain(str, task)
# In spawn
agent = await spawn(
premise="Fast responses",
model="openai:gpt-5" # Default: openai:gpt-4.1
)
# In decorator
@agentic(model="anthropic:claude-sonnet-4.5")
async def analyze(text: str) -> dict:
"""Analyze text."""
...
Available models:
openai:gpt-3.5-turbo, openai:gpt-4o, openai:gpt-4.1, openai:gpt-5anthropic:claude-sonnet-4, anthropic:claude-opus-4.1anthropic:claude-sonnet-4.5, anthropic:claude-opus-4.5google/gemini-2.5-flash)@agentic(persist=True)
async def chatbot(message: str) -> str:
"""Remembers conversation history."""
...
await chatbot("My name is Alice")
await chatbot("What's my name?") # Knows: Alice
For spawn() agents, state is automatic across calls to the same instance.
from agentica import spawn, MaxTokens
# Simple limit
agent = await spawn(
premise="Brief responses",
max_tokens=500
)
# Fine-grained control
agent = await spawn(
premise="Controlled output",
max_tokens=MaxTokens(
per_invocation=5000, # Total across all rounds
per_round=1000, # Per inference round
rounds=5 # Max inference rounds
)
)
from agentica import spawn, last_usage, total_usage
agent = await spawn(premise="You are helpful.")
await agent.call(str, "Hello!")
# Agent method
usage = agent.last_usage()
print(f"Last: {usage.input_tokens} in, {usage.output_tokens} out")
usage = agent.total_usage()
print(f"Total: {usage.total_tokens} processed")
# For @agentic functions
@agentic()
async def my_fn(x: str) -> str: ...
await my_fn("test")
print(last_usage(my_fn))
print(total_usage(my_fn))
from agentica import spawn
from agentica.logging.loggers import StreamLogger
import asyncio
agent = await spawn(premise="You are helpful.")
stream = StreamLogger()
with stream:
result = asyncio.create_task(
agent.call(bool, "Is Paris the capital of France?")
)
# Consume stream FIRST for live output
async for chunk in stream:
print(chunk.content, end="", flush=True)
# chunk.role is 'user', 'agent', or 'system'
# Then await result
final = await result
from agentica import spawn, agentic
# Via config file
agent = await spawn(
premise="Tool-using agent",
mcp="path/to/mcp_config.json"
)
@agentic(mcp="path/to/mcp_config.json")
async def tool_user(query: str) -> str:
"""Uses MCP tools."""
...
mcp_config.json format:
{
"mcpServers": {
"tavily-remote-mcp": {
"command": "npx -y mcp-remote https://mcp.tavily.com/mcp/?tavilyApiKey=<key>",
"env": {}
}
}
}
./logs/agent-<id>.logfrom agentica.logging.loggers import FileLogger, PrintLogger
from agentica.logging.agent_logger import NoLogging
# File only
with FileLogger():
agent = await spawn(premise="Debug agent")
await agent.call(int, "Calculate")
# Silent
with NoLogging():
agent = await spawn(premise="Silent agent")
# Listeners are in agent_listener submodule (NOT exported from agentica.logging)
from agentica.logging.agent_listener import (
PrintOnlyListener, # Console output only
FileOnlyListener, # File logging only
StandardListener, # Both console + file (default)
NoopListener, # Silent - no logging
)
agent = await spawn(
premise="Custom logging",
listener=PrintOnlyListener
)
# Silent agent
agent = await spawn(
premise="Silent agent",
listener=NoopListener
)
from agentica.logging.agent_listener import (
set_default_agent_listener,
get_default_agent_listener,
PrintOnlyListener,
)
set_default_agent_listener(PrintOnlyListener)
set_default_agent_listener(None) # Disable all
from agentica.errors import (
AgenticaError, # Base for all SDK errors
RateLimitError, # Rate limiting
InferenceError, # HTTP errors from inference
MaxTokensError, # Token limit exceeded
MaxRoundsError, # Max inference rounds exceeded
ContentFilteringError, # Content filtered
APIConnectionError, # Network issues
APITimeoutError, # Request timeout
InsufficientCreditsError,# Out of credits
OverloadedError, # Server overloaded
ServerError, # Generic server error
)
try:
result = await agent.call(str, "Do something")
except RateLimitError:
await asyncio.sleep(60)
result = await agent.call(str, "Do something")
except MaxTokensError:
# Reduce scope or increase limits
pass
except ContentFilteringError:
# Content was filtered
pass
except InferenceError as e:
logger.error(f"Inference failed: {e}")
except AgenticaError as e:
logger.error(f"SDK error: {e}")
class DataValidationError(Exception):
"""Invalid input data."""
pass
@agentic(DataValidationError) # Pass exception type
async def analyze(data: str) -> dict:
"""
Analyze data.
Raises:
DataValidationError: If data is malformed
"""
...
try:
result = await analyze(raw_data)
except DataValidationError as e:
logger.warning(f"Invalid: {e}")
from agentica.agent import Agent
class ResearchAgent:
def __init__(self, web_search_fn):
self._brain = Agent(
premise="Research assistant.",
scope={"web_search": web_search_fn}
)
async def research(self, topic: str) -> str:
return await self._brain(str, f"Research: {topic}")
async def summarize(self, text: str) -> str:
return await self._brain(str, f"Summarize: {text}")
class LeadResearcher:
def __init__(self):
self._brain = Agent(
premise="Coordinate research across subagents.",
scope={"SubAgent": ResearchAgent}
)
async def __call__(self, query: str) -> str:
return await self._brain(str, query)
lead = LeadResearcher()
report = await lead("Research AI agent frameworks 2025")
from agentica import initialize_tracing
# Initialize tracing (returns TracerProvider)
tracer = initialize_tracing(
service_name="my-agent-app",
environment="development", # Optional
tempo_endpoint="http://localhost:4317", # Optional: Grafana Tempo
organization_id="my-org", # Optional
log_level="INFO", # DEBUG, INFO, WARNING, ERROR
instrument_httpx=False, # Optional: trace HTTP calls
)
from agentica import enable_sdk_logging
# Enable internal SDK logs (for debugging the SDK itself)
disable_fn = enable_sdk_logging(log_tags="1")
# ... run agents ...
disable_fn() # Disable when done
# Main imports from agentica
from agentica import (
# Core
Agent, # Synchronous agent class
agentic, # @agentic decorator
spawn, # Async agent creation
# Configuration
ModelStrings, # Model string type hints
AgenticFunction, # Agentic function type
# Token tracking
last_usage, # Get last call's token usage
total_usage, # Get cumulative token usage
# Tracing/Logging
initialize_tracing, # OpenTelemetry setup
enable_sdk_logging, # SDK debug logs
# Version
__version__, # "0.3.1"
)
Before using Agentica:
@agentic() MUST be asyncspawn() returns awaitable - use await spawn(...)agent.call() is awaitable - use await agent.call(...)call() is return type, second is prompt stringpersist=True for conversation memory in @agenticAgent() (not spawn()) in synchronous Weekly Installs
198
Repository
GitHub Stars
3.6K
First Seen
Jan 22, 2026
Security Audits
Gen Agent Trust HubPassSocketPassSnykWarn
Installed on
opencode191
codex191
gemini-cli189
cursor187
github-copilot185
amp181
AI Elements:基于shadcn/ui的AI原生应用组件库,快速构建对话界面
62,200 周安装
__init__agentica.logging.agent_listener (NOT agentica.logging)