Composio 완전 가이드: AI Agent 에 1000개 이상의 도구 기능을 부여하는 오픈소스 프레임워크

# pip 사용
pip install composio

# 또는 poetry 사용
poetry add composio

pip install composio_openai_agents openai-agents

# npm 사용
npm install @composio/core

# yarn 사용
yarn add @composio/core

# pnpm 사용
pnpm add @composio/core

npm install @composio/openai-agents @openai/agents

export COMPOSIO_API_KEY="your-api-key-here"

import asyncio
from agents import Agent, Runner
from composio import Composio
from composio_openai_agents import OpenAIAgentsProvider

# Composio 클라이언트 초기화
composio = Composio(provider=OpenAIAgentsProvider())

# 사용자 ID 얻기 (인증 격리용)
user_id = "user@example.com"

# Hacker News 도구 얻기
tools = composio.tools.get(
    user_id=user_id,
    toolkits=["HACKERNEWS"]
)

# Agent 생성
agent = Agent(
    name="HackerNews Agent",
    instructions="당신은 Hacker News 어시스턴트입니다. 사용자에게 최신 기술 동향을 알려줄 수 있습니다.",
    tools=tools,
)

# Agent 실행
async def main():
    result = await Runner.run(
        starting_agent=agent,
        input="오늘 Hacker News에서 가장 인기 있는 기술 게시물은 무엇인가요?",
    )
    print(result.final_output)

asyncio.run(main())

import { Composio } from '@composio/core';
import { OpenAIAgentsProvider } from '@composio/openai-agents';
import { Agent, run } from '@openai/agents';

// SDK 초기화
const composio = new Composio({
  provider: new OpenAIAgentsProvider(),
});

const userId = 'user@example.com';

// 도구 얻기
const tools = await composio.tools.get(userId, {
  toolkits: ['HACKERNEWS'],
});

// Agent 생성
const agent = new Agent({
  name: 'HackerNews assistant',
  tools: tools,
});

// 실행
const result = await run(agent, 'What is the latest hackernews post about?');
console.log(JSON.stringify(result.finalOutput, null, 2));

from composio import Composio
from composio_openai_agents import OpenAIAgentsProvider

# 초기화
composio = Composio(provider=OpenAIAgentsProvider())
user_id = "dev-team@company.com"

# 여러 도구 패키지 얻기
tools = composio.tools.get(
    user_id=user_id,
    toolkits=["GITHUB", "SLACK"],
)

# DevOps Agent 생성
agent = Agent(
    name="DevOps Assistant",
    instructions="""
    당신은 DevOps 어시스턴트입니다. 다음과 같은 작업을 수행할 수 있습니다:
    1. GitHub 저장소의 Issue 상태 조회
    2. 새 Issue 또는 Pull Request 생성
    3. Slack 채널로 알림 전송

    중요한 Issue 업데이트가 발견되면 자동으로 팀에 알립니다.
    """,
    tools=tools,
)

async def monitor_issues():
    """핵심 Issue를 모니터링하고 팀에 알림"""
    result = await Runner.run(
        starting_agent=agent,
        input="""
        composiohq/composio 저장소에서 'bug' 라벨이 붙어 있고 7일 이상 응답이 없는 Issue를 확인하세요.
        목록을 #engineering Slack 채널로 보내주세요.
        """,
    )
    print(result.final_output)

# 인증 플로우를 트리거하는 단 한 줄의 코드
connection = composio.connections.initiate(
    user_id=user_id,
    app_name="gmail",
)
# 사용자가 OAuth 승인을 완료하면 Composio가 토큰을 자동으로 관리

from composio_langchain import ComposioToolSet

toolset = ComposioToolSet()
tools = toolset.get_tools(apps=["github", "slack"])

from langchain.agents import initialize_agent
agent = initialize_agent(
    tools=tools,
    llm=your_llm,
    agent_type="zero-shot-react-description"
)

from crewai import Agent, Task, Crew
from composio_crewai import ComposioToolSet

toolset = ComposioToolSet()
github_tools = toolset.get_tools(apps=["github"])

researcher = Agent(
    role='GitHub Researcher',
    goal='GitHub 트렌딩 프로젝트 분석',
    tools=github_tools,
    backstory='당신은 오픈소스 프로젝트 분석가입니다.',
)

from composio_autogen import ComposioToolSet

toolset = ComposioToolSet()
tools = toolset.get_tools(apps=["notion", "gmail"])

# AutoGen Agent에 도구 전달
assistant = ConversableAgent(
    "assistant",
    llm_config={"config_list": config_list},
    function_map=tools.get_function_map(),
)

from composio import Composio
from composio_openai_agents import OpenAIAgentsProvider

# 초기화
composio = Composio(provider=OpenAIAgentsProvider())
user_id = "support-bot@company.com"

# 고객 서비스 관련 도구 얻기
tools = composio.tools.get(
    user_id=user_id,
    toolkits=["NOTION", "JIRA", "GMAIL"],
)

# 고객 서비스 Agent 생성
support_agent = Agent(
    name="Customer Support Agent",
    instructions="""
    당신은 회사의 지능형 고객 서비스 어시스턴트입니다. 작업 흐름:

    1. 먼저 Notion 지식 베이스에서 사용자 질문의 답변을 검색
    2. 답변을 찾으면 Gmail로 사용자에게 발송
    3. 답변을 찾지 못하면 Jira에서 지원 티켓을 생성하고,
       기술 지원 팀에 할당한 후 Gmail로 사용자에게 티켓 번호를 알림

    친절하고 전문적인 어조를 유지하세요.
    """,
    tools=tools,
)

# 사용자 쿼리 처리
async def handle_customer_query(customer_email: str, question: str):
    prompt = f"""
    고객 이메일: {customer_email}
    질문: {question}

    작업 흐름에 따라 처리해주세요.
    """

    result = await Runner.run(
        starting_agent=support_agent,
        input=prompt,
    )
    return result.final_output

# 여러 데이터 소스를 동시에 조회
tools = composio.tools.get(
    user_id=user_id,
    toolkits=["GITHUB", "GITLAB", "BITBUCKET"],
)
# Agent는 세 플랫폼의 프로젝트 정보를 병렬로 조회할 수 있음

tools = composio.tools.get(
    user_id=user_id,
    toolkits=["HACKERNEWS"],
    cache_ttl=300,  # 5분 캐싱
)

# 개발 상황: 코드 관련 도구만 로드
dev_tools = composio.tools.get(user_id, toolkits=["GITHUB", "GITLAB"])

# 업무 상황: 생산성 도구만 로드
office_tools = composio.tools.get(user_id, toolkits=["GMAIL", "NOTION", "SLACK"])

try:
    result = await Runner.run(agent, input)
except ToolExecutionError as e:
    print(f"도구 실행 실패: {e}")
    # 대체 전략: 백업 도구 사용 또는 친절한 메시지 반환