diff --git a/app/core/engine.py b/app/core/engine.py index 882f165..959ac52 100644 --- a/app/core/engine.py +++ b/app/core/engine.py @@ -76,7 +76,7 @@ def agent_prompt(agent: Agent) -> str: prompt += """\n\nYou are integrated to Enso API, you are able to get the token list and their information, such as APY, Protocol Slug, Symbol, Address, and underlying tokens using enso_get_tokens tool. for each thread first request, you should use enso_get_tokens with no input param and get the information of the available protocol slugs, - symbols, addresses and APY.\n\n""" + symbols, addresses and APY. For token swap and route shortcut, you should use get_route tool.\n\n""" return prompt diff --git a/skills/enso/__init__.py b/skills/enso/__init__.py index 9869796..8092ff4 100644 --- a/skills/enso/__init__.py +++ b/skills/enso/__init__.py @@ -1,6 +1,7 @@ """Enso skills.""" from abstracts.skill import SkillStoreABC from skills.enso.base import EnsoBaseTool +from skills.enso.route import EnsoGetRouteShortcut from skills.enso.tokens import EnsoGetTokens @@ -12,6 +13,8 @@ def get_enso_skill( if name == "get_tokens": return EnsoGetTokens(api_token=api_token, main_tokens=main_tokens, store=store, agent_id=agent_id) + if name == "get_route_shortcut": + return EnsoGetRouteShortcut(api_token=api_token, main_tokens=main_tokens, store=store, agent_id=agent_id) else: raise ValueError(f"Unknown Enso skill: {name}") diff --git a/skills/enso/route.py b/skills/enso/route.py new file mode 100644 index 0000000..ca9ccd6 --- /dev/null +++ b/skills/enso/route.py @@ -0,0 +1,181 @@ +from typing import Literal, Type + +import httpx +from langchain_core.tools import ToolException +from pydantic import BaseModel, Field + +from .base import EnsoBaseTool, base_url + + +class EnsoGetRouteShortcutInput(BaseModel): + fromAddress: str = Field( + description="Ethereum address of the wallet to send the transaction from (It could be an EoA, or a Smart Wallet)." + ) + amountIn: list[int] = Field( + description="Amount of tokenIn to swap in wei." + ) + tokenIn: list[str] = Field( + description="Ethereum address of the token to swap or enter into a position from (For ETH, use 0xeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeee)." + ) + tokenOut: list[str] = Field( + description="Ethereum address of the token to swap or enter into a position to (For ETH, use 0xeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeee)." + ) + chainId: int | None = Field( + 1, + description="(Optional) Chain ID of the network to execute the transaction on." + ) + routingStrategy: Literal["router", "delegate", "ensowallet", None] = Field( + None, + description="(Optional) Routing strategy to use. Options: 'ensowallet', 'router', 'delegate'." + ) + receiver: str | None = Field( + None, + description="(Optional) Ethereum address of the receiver of the tokenOut." + ) + spender: str | None = Field( + None, + description="(Optional) Ethereum address of the spender of the tokenIn." + ) + amountOut: list[str] | None = Field( + None, + description="(Optional) Amount of tokenOut to receive." + ) + minAmountOut: list[str] | None = Field( + None, + description="(Optional) Minimum amount out in wei. If specified, slippage should not be specified." + ) + slippage: str | None = Field( + None, + description="(Optional) Slippage in basis points (1/10000). If specified, minAmountOut should not be specified." + ) + fee: list[str] | None = Field( + None, + description="(Optional) Fee in basis points (1/10000) for each amountIn value." + ) + feeReceiver: str | None = Field( + None, + description="(Optional) Ethereum address that will receive the collected fee if fee was provided." + ) + disableRFQs: bool | None = Field( + None, + description="(Optional) Exclude RFQ sources from routes." + ) + ignoreAggregators: list[str] | None = Field( + None, + description="(Optional) List of swap aggregators to ignore." + ) + ignoreStandards: list[str] | None = Field( + None, + description="(Optional) List of standards to ignore." + ) + variableEstimates: dict | None = Field( + None, + description="Variable estimates for the route." + ) + + +class Transaction(BaseModel): + data: str = Field(None, description="Data of the transaction.") + to: str = Field(None, description="Ethereum address of the receiver of the transaction.") + from_: str = Field(None, description="Ethereum address of the sender of the transaction.") + value: str = Field(None, description="Amount of token to send.") + + +class Route(BaseModel): + tokenIn: list[str] | None = Field( + None, description="Ethereum address of the token to swap or enter into a position from." + ) + tokenOut: list[str] | None = Field( + None, description="Ethereum address of the token to swap or enter into a position to." + ) + protocol: str | None = Field(None, description="Protocol used for finding route.") + action: str | None = Field(None, description="Action has been done for route (e.g. swap).") + internalRoutes: list[str] | None = Field(None, description="Internal routes needed for the route.") + + +class RouteShortcutGetTransaction(BaseModel): + """ + Output schema for the `/api/v1/shortcuts/route` GET endpoint. + """ + gas: str | None = Field(None, description="Gas amount for the transaction.") + amountOut: str | dict | None = Field(None, description="The final calculated amountOut as an object.") + priceImpact: float | None = Field(None, + description="Price impact in basis points, it is null if USD price is not found.") + feeAmount: list[str] | None = Field(None, description="An array of the fee amounts collected for each tokenIn.") + createdAt: int | None = Field(None, description="Block number the transaction was created on.") + tx: Transaction | None = Field(None, description="The tx object to use in `ethers`.") + route: list[Route] | None = Field(None, description="Route that the shortcut will use.") + + +class EnsoGetRouteShortcutOutput(BaseModel): + res: RouteShortcutGetTransaction | None = None + error: str | None = None + + +class EnsoGetRouteShortcut(EnsoBaseTool): + """ + This tool finds the optimal execution path across a multitude of DeFi protocols such as liquidity pools, + lending platforms, automated market makers, yield optimizers, and more. This allows for maximized capital efficiency + and yield optimization, taking into account return rates, gas costs, and slippage. + + Attributes: + name (str): Name of the tool, specifically "enso_get_route_shortcut". + description (str): Comprehensive description of the tool's purpose and functionality. + args_schema (Type[BaseModel]): Schema for input arguments, specifying expected parameters. + """ + + name: str = "enso_get_route_shortcut" + description: str = "This tool optimizes DeFi swaps, identifying the most efficient execution path across various protocols (e.g., liquidity pools, lending platforms) by considering factors like return rates, gas costs, and slippage." + args_schema: Type[BaseModel] = EnsoGetRouteShortcutInput + + def _run(self, fromAddress: str, amountIn: list[int], tokenIn: list[str], tokenOut: list[str], + **kwargs) -> EnsoGetRouteShortcutOutput: + """ + Run the tool to get swap route information. + + Args: + fromAddress (str): Ethereum address of the wallet to send the transaction from (It could be an EoA, or a Smart Wallet). + amountIn (list[int]): Amount of tokenIn to swap in wei. + tokenIn (list[str]): Ethereum address of the token to swap or enter into a position from (For ETH, use 0xeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeee). + tokenOut (list[str]): Ethereum address of the token to swap or enter into a position to (For ETH, use 0xeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeee). + **kwargs: kwargs optional arguments for the tool with args schema defined in EnsoGetTokensInput. + + Returns: + EnsoGetRouteShortcutOutput: The response containing route shortcut information. + """ + url = f"{base_url}/api/v1/shortcuts/route" + + headers = { + "Authorization": f"Bearer {self.api_token}", + } + + # Prepare query parameters + params = EnsoGetRouteShortcutInput(fromAddress=fromAddress, amountIn=amountIn, tokenIn=tokenIn, + tokenOut=tokenOut, **kwargs) + + if params.fromAddress == "0xYourWalletAddress": + raise ToolException("the ethereum from address is required") + + with httpx.Client() as client: + try: + response = client.get(url, headers=headers, params=params.model_dump(exclude_none=True)) + response.raise_for_status() # Raise HTTPError for non-2xx responses + + # Parse and return the response as a RouteShortcutGetTransaction object + return EnsoGetRouteShortcutOutput(res=RouteShortcutGetTransaction(**response.json()), error=None) + + except httpx.RequestError as req_err: + return EnsoGetRouteShortcutOutput(res=None, error=f"Request error: {req_err}") + except httpx.HTTPStatusError as http_err: + return EnsoGetRouteShortcutOutput(res=None, error=f"HTTP error: {http_err}") + except Exception as e: + return EnsoGetRouteShortcutOutput(res=None, error=str(e)) + + async def _arun(self, fromAddress: str, amountIn: list[int], tokenIn: list[str], tokenOut: list[str], + **kwargs) -> EnsoGetRouteShortcutOutput: + """Async implementation of the tool. + + This tool doesn't have a native async implementation, so we call the sync version. + """ + return self._run(fromAddress=fromAddress, amountIn=amountIn, tokenIn=tokenIn, + tokenOut=tokenOut, **kwargs)