Python custom provider

1
from typing import List, Optional, Sequence, TypeAlias
2
from composio.core.provider import NonAgenticProvider
3
from composio.types import Tool, Modifiers, ToolExecutionResponse
4
5
# Define your tool format
6
class MyAITool:
7
    def __init__(self, name: str, description: str, parameters: dict):
8
        self.name = name
9
        self.description = description
10
        self.parameters = parameters
11
12
# Define your tool collection format
13
MyAIToolCollection: TypeAlias = List[MyAITool]
14
15
# Create your provider
16
class MyAIProvider(NonAgenticProvider[MyAITool, MyAIToolCollection], name="my-ai-platform"):
17
    """Custom provider for My AI Platform"""
18
    
19
    def wrap_tool(self, tool: Tool) -> MyAITool:
20
        """Transform a single tool to platform format"""
21
        return MyAITool(
22
            name=tool.slug,
23
            description=tool.description or "",
24
            parameters={
25
                "type": "object",
26
                "properties": tool.input_parameters.get("properties", {}),
27
                "required": tool.input_parameters.get("required", [])
28
            }
29
        )
30
    
31
    def wrap_tools(self, tools: Sequence[Tool]) -> MyAIToolCollection:
32
        """Transform a collection of tools"""
33
        return [self.wrap_tool(tool) for tool in tools]
34
    
35
    # Optional: Custom helper methods for your AI platform
36
    def execute_my_ai_tool_call(
37
        self,
38
        user_id: str,
39
        tool_call: dict,
40
        modifiers: Optional[Modifiers] = None
41
    ) -> ToolExecutionResponse:
42
        """Execute a tool call in your platform's format
43
        
44
        Example usage:
45
        result = my_provider.execute_my_ai_tool_call(
46
            user_id="default",
47
            tool_call={"name": "GITHUB_STAR_REPO", "arguments": {"owner": "composiohq", "repo": "composio"}}
48
        )
49
        """
50
        # Use the built-in execute_tool method
51
        return self.execute_tool(
52
            slug=tool_call["name"],
53
            arguments=tool_call["arguments"],
54
            modifiers=modifiers,
55
            user_id=user_id
56
        )

1
from typing import Callable, Dict, List, Sequence
2
from composio.core.provider import AgenticProvider, AgenticProviderExecuteFn
3
from composio.types import Tool
4
from my_provider import AgentTool
5
6
# Import the Tool/Function class that represents a callable tool for your framework
7
# Optionally define your custom tool format below
8
# class AgentTool:
9
#     def __init__(self, name: str, description: str, execute: Callable, schema: dict):
10
#         self.name = name
11
#         self.description = description
12
#         self.execute = execute
13
#         self.schema = schema
14
15
# Define your tool collection format (typically a List)
16
AgentToolCollection: TypeAlias = List[AgentTool]
17
18
# Create your provider
19
class MyAgentProvider(AgenticProvider[AgentTool, List[AgentTool]], name="my-agent-platform"):
20
    """Custom provider for My Agent Platform"""
21
    
22
    def wrap_tool(self, tool: Tool, execute_tool: AgenticProviderExecuteFn) -> AgentTool:
23
        """Transform a single tool with execute function"""
24
        def execute_wrapper(**kwargs) -> Dict:
25
            result = execute_tool(tool.slug, kwargs)
26
            if not result.get("successful", False):
27
                raise Exception(result.get("error", "Tool execution failed"))
28
            return result.get("data", {})
29
        
30
        return AgentTool(
31
            name=tool.slug,
32
            description=tool.description or "",
33
            execute=execute_wrapper,
34
            schema=tool.input_parameters
35
        )
36
    
37
    def wrap_tools(
38
        self, 
39
        tools: Sequence[Tool], 
40
        execute_tool: AgenticProviderExecuteFn
41
    ) -> AgentToolCollection:
42
        """Transform a collection of tools with execute function"""
43
        return [self.wrap_tool(tool, execute_tool) for tool in tools]

1
from composio import Composio
2
from composio_myai import MyAIProvider
3
from myai import MyAIClient  # Your AI platform's client
4
5
# Initialize tools
6
myai_client = MyAIClient()
7
composio = Composio(provider=MyAIProvider())
8
9
# Define task
10
task = "Star a repo composiohq/composio on GitHub"
11
12
# Get GitHub tools that are pre-configured
13
tools = composio.tools.get(user_id="default", toolkits=["GITHUB"])
14
15
# Get response from your AI platform (example)
16
response = myai_client.chat.completions.create(
17
    model="your-model",
18
    tools=tools,  # These are in your platform's format
19
    messages=[
20
        {"role": "system", "content": "You are a helpful assistant."},
21
        {"role": "user", "content": task},
22
    ],
23
)
24
print(response)
25
26
# Execute the function calls
27
result = composio.provider.handle_tool_calls(response=response, user_id="default")
28
print(result)

1
import asyncio
2
from agents import Agent, Runner
3
from composio_myagent import MyAgentProvider
4
from composio import Composio
5
6
# Initialize Composio toolset
7
composio = Composio(provider=MyAgentProvider())
8
9
# Get all the tools
10
tools = composio.tools.get(
11
    user_id="default",
12
    tools=["GITHUB_STAR_A_REPOSITORY_FOR_THE_AUTHENTICATED_USER"],
13
)
14
15
# Create an agent with the tools
16
agent = Agent(
17
    name="GitHub Agent",
18
    instructions="You are a helpful assistant that helps users with GitHub tasks.",
19
    tools=tools,
20
)
21
22
# Run the agent
23
async def main():
24
    result = await Runner.run(
25
        starting_agent=agent,
26
        input=(
27
            "Star the repository composiohq/composio on GitHub. If done "
28
            "successfully, respond with 'Action executed successfully'"
29
        ),
30
    )
31
    print(result.final_output)
32
33
asyncio.run(main())

1
import typing as t
2
3
from anthropic.types.beta.beta_tool_use_block import BetaToolUseBlock
4
from anthropic.types.message import Message as ToolsBetaMessage
5
from anthropic.types.tool_param import ToolParam
6
from anthropic.types.tool_use_block import ToolUseBlock
7
8
from composio.core.provider import NonAgenticProvider
9
from composio.types import Modifiers, Tool, ToolExecutionResponse
10
11
12
class AnthropicProvider(
13
    NonAgenticProvider[ToolParam, list[ToolParam]],
14
    name="anthropic",
15
):
16
    """Composio toolset for Anthropic Claude platform."""
17
18
    def wrap_tool(self, tool: Tool) -> ToolParam:
19
        return ToolParam(
20
            input_schema=tool.input_parameters,
21
            name=tool.slug,
22
            description=tool.description,
23
        )
24
25
    def wrap_tools(self, tools: t.Sequence[Tool]) -> list[ToolParam]:
26
        return [self.wrap_tool(tool) for tool in tools]
27
28
    def execute_tool_call(
29
        self,
30
        user_id: str,
31
        tool_call: ToolUseBlock,
32
        modifiers: t.Optional[Modifiers] = None,
33
    ) -> ToolExecutionResponse:
34
        """Execute a tool call.
35
        
36
        :param user_id: User ID to use for executing function calls.
37
        :param tool_call: Tool call metadata.
38
        :param modifiers: Modifiers to use for executing function calls.
39
        :return: Object containing output data from the tool call.
40
        """
41
        return self.execute_tool(
42
            slug=tool_call.name,
43
            arguments=t.cast(t.Dict, tool_call.input),
44
            modifiers=modifiers,
45
            user_id=user_id,
46
        )
47
48
    def handle_tool_calls(
49
        self,
50
        user_id: str,
51
        response: t.Union[dict, ToolsBetaMessage],
52
        modifiers: t.Optional[Modifiers] = None,
53
    ) -> t.List[ToolExecutionResponse]:
54
        """Handle tool calls from Anthropic Claude chat completion object.
55
        
56
        :param response: Chat completion object from
57
            `anthropic.Anthropic.beta.tools.messages.create` function call.
58
        :param user_id: User ID to use for executing function calls.
59
        :param modifiers: Modifiers to use for executing function calls.
60
        :return: A list of output objects from the tool calls.
61
        """
62
        if isinstance(response, dict):
63
            response = ToolsBetaMessage(**response)
64
65
        outputs = []
66
        for content in response.content:
67
            if isinstance(content, (ToolUseBlock, BetaToolUseBlock)):
68
                outputs.append(
69
                    self.execute_tool_call(
70
                        user_id=user_id,
71
                        tool_call=content,
72
                        modifiers=modifiers,
73
                    )
74
                )
75
        return outputs

1
import asyncio
2
import json
3
import typing as t
4
5
import pydantic
6
from agents import FunctionTool
7
8
from composio.core.provider import AgenticProvider, AgenticProviderExecuteFn
9
from composio.types import Tool
10
from composio.utils.pydantic import parse_pydantic_error
11
12
13
class OpenAIAgentsProvider(
14
    AgenticProvider[FunctionTool, list[FunctionTool]],
15
    name="openai_agents",
16
):
17
    """Provider for OpenAI Agents framework"""
18
    
19
    def wrap_tool(
20
        self,
21
        tool: Tool,
22
        execute_tool: AgenticProviderExecuteFn,
23
    ) -> FunctionTool:
24
        """Wrap a tool as a FunctionTool"""
25
        
26
        # Create an async wrapper for the tool execution
27
        async def execute_tool_wrapper(_ctx, payload):
28
            """Execute Composio action with the given arguments"""
29
            try:
30
                return json.dumps(
31
                    obj=(
32
                        await asyncio.to_thread(  # Running in thread since execute_tool is not async
33
                            execute_tool,
34
                            slug=tool.slug,
35
                            arguments=json.loads(payload) if payload else {},
36
                        )
37
                    )
38
                )
39
            except pydantic.ValidationError as e:
40
                return json.dumps({
41
                    "successful": False,
42
                    "error": parse_pydantic_error(e),
43
                    "data": None,
44
                })
45
            except Exception as e:
46
                return json.dumps({
47
                    "successful": False,
48
                    "error": str(e),
49
                    "data": None,
50
                })
51
        
52
        # Ensure the schema has additionalProperties set to false
53
        modified_schema = tool.input_parameters.copy()
54
        modified_schema["additionalProperties"] = False
55
        
56
        # Create a FunctionTool with the appropriate schema
57
        return FunctionTool(
58
            name=tool.slug,
59
            description=tool.description,
60
            params_json_schema=modified_schema,
61
            on_invoke_tool=execute_tool_wrapper,
62
            strict_json_schema=False,  # Composio tools already have optimal schemas
63
        )
64
    
65
    def wrap_tools(
66
        self,
67
        tools: t.Sequence[Tool],
68
        execute_tool: AgenticProviderExecuteFn,
69
    ) -> list[FunctionTool]:
70
        """Wrap multiple tools for OpenAI Agents"""
71
        return [self.wrap_tool(tool, execute_tool) for tool in tools]