mcpField
P
python-sdk

python-sdk

github.com/inference-sh开发7891.1k 次安装2 周前更新
代码仓库

关于 python-sdk

用于运行 inference.sh 应用并构建 Agent 的 Python SDK,支持工具、会话和流式响应,适合将应用执行或 Agent 工作流集成到 Python 项目的开发者。

主要功能

  1. 01同步或异步运行应用
  2. 02跟踪任务并流式接收进度
  3. 03上传文件并管理会话
  4. 04构建 Agent 和工具定义
  5. 05支持审批工作流

适用场景

  1. 01将 AI 应用运行集成到 Python 服务
  2. 02创建可调用应用或委派任务的 Agent
  3. 03通过应用输入处理文件
  4. 04展示长任务的执行进度

安装

复制下面的命令,在终端中运行。

$npx skills add https://github.com/inference-sh/skills --skill python-sdk

安装后在 Claude Code 中输入/python-sdk确认已加载。

包含的文件

除 SKILL.md 外,安装时一起下载的参考资料和脚本。

references/
agent-patterns.md
async-patterns.md
files.md
sessions.md
streaming.md
tool-builder.md

详细介绍

安装 belt CLI 技能: npx skills add belt-sh/cli

Python SDK

使用 inference.sh Python SDK 构建 AI 应用。

Python SDK

快速开始

pip install inferencesh
from inferencesh import inference

client = inference(api_key="inf_your_key")

# Run an AI app
result = client.run({
    "app": "infsh/flux-1-dev",
    "input": {"prompt": "A sunset over mountains"}
})
print(result["output"])

安装

# Standard installation
pip install inferencesh

# With async support
pip install inferencesh[async]

要求: Python 3.8+

身份验证

import os
from inferencesh import inference

# Direct API key
client = inference(api_key="inf_your_key")

# From environment variable (recommended)
client = inference(api_key=os.environ["INFERENCE_API_KEY"])

获取 API 密钥:Settings → API Keys → Create API Key

运行应用

基本执行
result = client.run({
    "app": "infsh/flux-1-dev",
    "input": {"prompt": "A cat astronaut"}
})

print(result["status"])  # "completed"
print(result["output"])  # Output data
触发后即不再等待
task = client.run({
    "app": "google/veo-3-1-fast",
    "input": {"prompt": "Drone flying over mountains"}
}, wait=False)

print(f"Task ID: {task['id']}")
# Check later with client.get_task(task['id'])
流式进度
for update in client.run({
    "app": "google/veo-3-1-fast",
    "input": {"prompt": "Ocean waves at sunset"}
}, stream=True):
    print(f"Status: {update['status']}")
    if update.get("logs"):
        print(update["logs"][-1])
运行参数
参数类型说明
appstringApp ID(namespace/name@version)
inputdict与 app 架构匹配的输入
setupdict隐藏的设置配置
infrastring'cloud' 或 'private'
sessionstring有状态执行的会话 ID
session_timeoutint空闲超时时间(1-3600 秒)

文件处理

自动上传
result = client.run({
    "app": "image-processor",
    "input": {
        "image": "/path/to/image.png"  # Auto-uploaded
    }
})
手动上传
from inferencesh import UploadFileOptions

# Basic upload
file = client.upload_file("/path/to/image.png")

# With options
file = client.upload_file(
    "/path/to/image.png",
    UploadFileOptions(
        filename="custom_name.png",
        content_type="image/png",
        public=True
    )
)

result = client.run({
    "app": "image-processor",
    "input": {"image": file["uri"]}
})

会话(有状态执行)

在多次调用之间保持工作进程处于热状态:

# Start new session
result = client.run({
    "app": "my-app",
    "input": {"action": "init"},
    "session": "new",
    "session_timeout": 300  # 5 minutes
})
session_id = result["session_id"]

# Continue in same session
result = client.run({
    "app": "my-app",
    "input": {"action": "process"},
    "session": session_id
})

Agent SDK

模板 Agent

使用工作区中预先构建的 Agent:

agent = client.agent("my-team/support-agent@latest")

# Send message
response = agent.send_message("Hello!")
print(response.text)

# Multi-turn conversation
response = agent.send_message("Tell me more")

# Reset conversation
agent.reset()

# Get chat history
chat = agent.get_chat()
临时 Agent

以编程方式创建自定义 Agent:

from inferencesh import tool, string, number, app_tool

# Define tools
calculator = (
    tool("calculate")
    .describe("Perform a calculation")
    .param("expression", string("Math expression"))
    .build()
)

image_gen = (
    app_tool("generate_image", "infsh/flux-1-dev@latest")
    .describe("Generate an image")
    .param("prompt", string("Image description"))
    .build()
)

# Create agent
agent = client.agent({
    "core_app": {"ref": "infsh/claude-sonnet-4@latest"},
    "system_prompt": "You are a helpful assistant.",
    "tools": [calculator, image_gen],
    "temperature": 0.7,
    "max_tokens": 4096
})

response = agent.send_message("What is 25 * 4?")
可用的核心应用
模型应用引用
Claude Sonnet 4infsh/claude-sonnet-4@latest
Claude 3.5 Haikuinfsh/claude-haiku-35@latest
GPT-4oinfsh/gpt-4o@latest
GPT-4o Miniinfsh/gpt-4o-mini@latest

工具构建器 API

参数类型
from inferencesh import (
    string, number, integer, boolean,
    enum_of, array, obj, optional
)

name = string("User's name")
age = integer("Age in years")
score = number("Score 0-1")
active = boolean("Is active")
priority = enum_of(["low", "medium", "high"], "Priority")
tags = array(string("Tag"), "List of tags")
address = obj({
    "street": string("Street"),
    "city": string("City"),
    "zip": optional(string("ZIP"))
}, "Address")
客户端工具(在你的代码中运行)
greet = (
    tool("greet")
    .display("Greet User")
    .describe("Greets a user by name")
    .param("name", string("Name to greet"))
    .require_approval()
    .build()
)
应用工具(调用 AI 应用)
generate = (
    app_tool("generate_image", "infsh/flux-1-dev@latest")
    .describe("Generate an image from text")
    .param("prompt", string("Image description"))
    .setup({"model": "schnell"})
    .input({"steps": 20})
    .require_approval()
    .build()
)
Agent 工具(委派给子 Agent)
from inferencesh import agent_tool

researcher = (
    agent_tool("research", "my-org/researcher@v1")
    .describe("Research a topic")
    .param("topic", string("Topic to research"))
    .build()
)
Webhook 工具(调用外部 API)
from inferencesh import webhook_tool

notify = (
    webhook_tool("slack", "https://hooks.slack.com/...")
    .describe("Send Slack notification")
    .secret("SLACK_SECRET")
    .param("channel", string("Channel"))
    .param("message", string("Message"))
    .build()
)
内部工具(内置功能)
from inferencesh import internal_tools

config = (
    internal_tools()
    .plan()
    .memory()
    .web_search(True)
    .code_execution(True)
    .image_generation({
        "enabled": True,
        "app_ref": "infsh/flux@latest"
    })
    .build()
)

agent = client.agent({
    "core_app": {"ref": "infsh/claude-sonnet-4@latest"},
    "internal_tools": config
})

流式 Agent 响应

def handle_message(msg):
    if msg.get("content"):
        print(msg["content"], end="", flush=True)

def handle_tool(call):
    print(f"\n[Tool: {call.name}]")
    result = execute_tool(call.name, call.args)
    agent.submit_tool_result(call.id, result)

response = agent.send_message(
    "Explain quantum computing",
    on_message=handle_message,
    on_tool_call=handle_tool
)

文件附件

# From file path
with open("image.png", "rb") as f:
    response = agent.send_message(
        "What's in this image?",
        files=[f.read()]
    )

# From base64
response = agent.send_message(
    "Analyze this",
    files=["data:image/png;base64,iVBORw0KGgo..."]
)

技能(可复用上下文)

agent = client.agent({
    "core_app": {"ref": "infsh/claude-sonnet-4@latest"},
    "skills": [
        {
            "name": "code-review",
            "description": "Code review guidelines",
            "content": "# Code Review\n\n1. Check security\n2. Check performance..."
        },
        {
            "name": "api-docs",
            "description": "API documentation",
            "url": "https://example.com/skills/api-docs.md"
        }
    ]
})

异步支持

from inferencesh import async_inference
import asyncio

async def main():
    client = async_inference(api_key="inf_...")

    # Async app execution
    result = await client.run({
        "app": "infsh/flux-1-dev",
        "input": {"prompt": "A galaxy"}
    })

    # Async agent
    agent = client.agent("my-org/assistant@latest")
    response = await agent.send_message("Hello!")

    # Async streaming
    async for msg in agent.stream_messages():
        print(msg)

asyncio.run(main())

错误处理

from inferencesh import RequirementsNotMetException

try:
    result = client.run({"app": "my-app", "input": {...}})
except RequirementsNotMetException as e:
    print(f"Missing requirements:")
    for err in e.errors:
        print(f"  - {err['type']}: {err['key']}")
except RuntimeError as e:
    print(f"Error: {e}")

人工审批工作流

def handle_tool(call):
    if call.requires_approval:
        # Show to user, get confirmation
        approved = prompt_user(f"Allow {call.name}?")
        if approved:
            result = execute_tool(call.name, call.args)
            agent.submit_tool_result(call.id, result)
        else:
            agent.submit_tool_result(call.id, {"error": "Denied by user"})

response = agent.send_message(
    "Delete all temp files",
    on_tool_call=handle_tool
)

参考文件

相关技能

# JavaScript SDK
npx skills add inference-sh/skills@javascript-sdk

# Full platform skill (all apps via CLI)
npx skills add inference-sh/skills@infsh-cli

# LLM models
npx skills add inference-sh/skills@llm-models

# Image generation
npx skills add inference-sh/skills@ai-image-generation

文档

评论

支持 Markdown