ACI.dev - 开源 AI 代理工具集成平台

   git clone https://github.com/aipotheosis-labs/aci.git   cd aci

   pip install -r requirements.txt

   docker-compose up -d

   alembic upgrade head

from aci.cli.commands import create_project, create_agent
project = create_project(name="My Project", org_id="your-org-id")agent = create_agent(project_id=project.id, name="My Agent")

import aci
# 初始化SDKclient = aci.Client(api_key="your-api-key")
# 搜索功能functions = client.search_functions(intent="发送邮件")
# 执行功能result = client.execute_function(    function_name="GMAIL__SEND_EMAIL",    arguments={"to": "user@example.com", "subject": "Hello", "body": "Test"})

"""创建App表迁移"""from alembic import opimport sqlalchemy as sa
def upgrade():    op.create_table(        "apps",        sa.Column("id", sa.UUID(), nullable=False),        sa.Column("name", sa.String(length=100), nullable=False),        sa.Column("display_name", sa.String(length=255), nullable=False),        sa.Column("provider", sa.String(length=255), nullable=False),        sa.Column("version", sa.String(length=255), nullable=False),        sa.Column("description", sa.Text(), nullable=False),        sa.Column("logo", sa.Text(), nullable=True),        sa.Column("categories", sa.ARRAY(sa.String()), nullable=False),        sa.Column("visibility", sa.Enum("PUBLIC", "PRIVATE"), nullable=False),        sa.PrimaryKeyConstraint("id"),        sa.UniqueConstraint("name")    )

class FunctionExecutor:    """功能执行器基类"""        def execute(self, function, function_input):        # 预处理输入        function_input = self._preprocess_input(function, function_input)                # 执行功能        return self._execute(function, function_input)
    def _preprocess_input(self, function, input_data):        # 验证输入schema        jsonschema.validate(input_data, function.parameters)                # 注入默认值        return processor.inject_defaults(function.parameters, input_data)

class OAuth2Manager:    """OAuth2流程管理器"""        def __init__(self, client_id, client_secret, auth_url, token_url):        self.client = AsyncOAuth2Client(            client_id=client_id,            client_secret=client_secret,            authorize_url=auth_url,            token_url=token_url        )        async def get_token(self, code):        return await self.client.fetch_token(code=code)