feat: add MCP server to uni CLI

.github/workflows/test.yaml

@@ -15,7 +15,7 @@ jobs:
             python-version: "3.10"
 
         - name: Install Dependencies
-          run: pip install .[lint,bot]
+          run: pip install .[lint,bot,mcp]
 
         - name: Run Black
           run: black --check .
@@ -38,7 +38,7 @@ jobs:
             python-version: "3.10"
 
         - name: Install Dependencies
-          run: pip install .[lint,test,bot]  # Might need test deps
+          run: pip install .[lint,test,bot,mcp]  # Might need test deps
 
         - name: Run MyPy
           run: mypy .

README.md

@@ -72,6 +72,64 @@ You can use this command to do common tasks with the SDK such as finding prices
 
 Try `uni --help` after installing the SDK to learn more about what the CLI can do.
 
+### MCP Tool
+
+If you want to use this SDK w/ an LLM that supports tool calling via MCP, there is a CLI method for that!
+
+First off, you need to install Ape, relevant Ape plugins (such as Wallets, Explorers, Data & RPC Providers, etc.).
+Then, you should configure wallets for use in Ape (see the [Ape docs](https://docs.apeworx.io/ape/latest/userguides/accounts#live-network-accounts) on setting up a wallet for live network use).
+Finally, install this SDK and launch the MCP tool via:
+
+```sh
+uni mcp --network ... --account ... --token WETH --token ...
+```
+
+This will launch the MCP server completely locally (connected to your local accounts).
+Configure your preferred LLM to use this MCP tool via config.
+
+Claude-style Config file:
+
+```json
+  ...
+  "mcpServers": {
+    ...
+    "Uniswap": {
+      "command": "/home/doggie/.local/bin/uvx",
+      "args": [
+        "--from",
+        "uniswap-sdk[mcp]",
+        // Plugins you need
+        "--with",
+        "ape-base",
+        "--with",
+        "ape-frame",
+        "uni",
+        "mcp",
+        // Tokens you want to index
+        "--token",
+        "WETH",
+        "--network",
+        // Network you want to use (include provider!)
+        "base:mainnet:node",
+        // Account you want to use for checking balances & trading with
+        "--account",
+        "frame"
+      ]
+    },
+    },
+    ...
+  }
+  ...
+```
+
+Then prompt!
+
+```{notice}
+Sending swaps via MCP can be very dangerous if you don't know what you're doing,
+however thanks to how the MCP server functions you will need to approve every transaction it initiates.
+Take care to verify each transaction to ensure that the tool call was successfully translated.
+```
+
 ### Silverback
 
 The SDK has special support for use within [Silverback](https://silverback.apeworx.io) bots,

setup.py

@@ -30,6 +30,9 @@
     "bot": [
         "silverback>=0.7.22",  # Need for linting as it is optional
     ],
+    "mcp": [
+        "fastmcp>=2.10,<3",  # Needed for `uni mcp` server
+    ],
 }
 
 # NOTE: `pip install -e .[dev]` to install package
@@ -38,6 +41,7 @@
     + extras_require["lint"]
     + extras_require["release"]
     + extras_require["bot"]
+    + extras_require["mcp"]
     + extras_require["dev"]
 )
 

uniswap_sdk/_cli.py

@@ -1,8 +1,12 @@
+from contextlib import asynccontextmanager
 from decimal import Decimal
+from typing import Annotated
 
 import click
-from ape.cli import ConnectedProviderCommand, account_option, network_option
+from ape.cli import ConnectedProviderCommand, account_option, network_option, verbosity_option
+from ape.types import AddressType
 from ape_tokens import Token, tokens
+from pydantic import Field
 
 from uniswap_sdk import Uniswap
 
@@ -19,7 +23,7 @@ def intermediate_tokens():
         metavar="TOKEN",
         multiple=True,
         default=[],
-        help="Tokens to index intermediate routes for.",
+        help="Tokens to index routes for.",
     )
 
 
@@ -131,3 +135,170 @@ def swap(
         receiver=receiver,
         sender=account,
     )
+
+
+@cli.command(cls=ConnectedProviderCommand)
+@verbosity_option(default=100_000)  # NOTE: Disabled
+@network_option()
+@account_option()
+@intermediate_tokens()
+def mcp(ecosystem, network, filter_tokens, account):
+    """Start the Uniswap MCP Server"""
+
+    try:
+        from fastmcp import Context, FastMCP
+
+    except ImportError:
+        raise click.UsageError("Must install the `[mcp]` extra to use this command.")
+
+    @asynccontextmanager
+    async def lifespan(server):
+        uni = Uniswap()
+        list(uni.index(tokens=(filter_tokens or tokens)))
+        yield uni
+
+    server = FastMCP(
+        name=f"Uniswap Protocol on {ecosystem.name}:{network.name}",
+        lifespan=lifespan,
+        instructions=f"""
+        # Uniswap MCP Server
+
+        This server provides capabilities for pricing and swapping tokens
+        using the Uniswap protocol on {ecosystem.name}:{network.name}
+        for the user account {account.address}.
+        """,
+    )
+
+    # TODO: Move this to ape-tokens?
+    @server.tool()
+    async def get_token_balance(
+        token: Annotated[str | AddressType, Field(description="The token symbol or address")],
+    ) -> Decimal:
+        """Get the balance of `token` in the user's account."""
+
+        if token in ("ether", "ETH"):
+            return account.balance * Decimal("1e-18")
+
+        from ape import convert
+        from ape.types import AddressType
+
+        token = Token.at(convert(token, AddressType))
+        return token.balanceOf(account) * Decimal(  # type: ignore[attr-defined]
+            f"1e-{token.decimals()}"  # type: ignore[attr-defined]
+        )
+
+    @server.tool()
+    async def get_price(
+        ctx: Context,
+        base: Annotated[
+            str | AddressType,
+            Field(description="The token symbol or address you want to know the price of"),
+        ],
+        quote: Annotated[
+            str | AddressType,
+            Field(description="The token symbol or address which the price will be expressed"),
+        ],
+    ) -> Decimal:
+        """
+        Returns the current exchange rate between two tokens, `base` and `quote`, as observed
+        across all relevant markets in the Uniswap protocol. This price reflects the starting rate
+        at which a trade on Uniswap will begin, and it does not include slippage or market impact
+        from conducting an actual trade. due to the mechanics of the Uniswap AMM model.
+
+        **Important Notes**:
+        1. It is only intended to use this price as a reference.
+        2. The number will be returned as a decimal value, reflecting the precision of the market
+           price. Do not scale or re-interpret this number.
+        3. The number should be interpretted as being the number of `quote` tokens that equals
+           exactly 1 `base` token by the current market, or in the context of `quote` per `base`.
+        """
+
+        uni = ctx.request_context.lifespan_context
+        return uni.price(base, quote)
+
+    @server.tool()
+    async def swap(
+        ctx: Context,
+        have: Annotated[
+            str | AddressType,
+            Field(description="The token symbol or address you want to sell"),
+        ],
+        want: Annotated[
+            str | AddressType,
+            Field(description="The token symbol or address you want to buy"),
+        ],
+        amount_in: Annotated[
+            Decimal | None,
+            Field(
+                description="The amount of `have` tokens you want to sell."
+                " Leave empty if using `amount_out`."
+            ),
+        ] = None,
+        max_amount_in: Annotated[
+            Decimal | None,
+            Field(
+                description="The maximum amount of `have` tokens you are willing to sell."
+                " Leave empty to auto-compute this amount when `amount_out` is provided."
+            ),
+        ] = None,
+        amount_out: Annotated[
+            Decimal | None,
+            Field(
+                description="The amount of `want` tokens you want to buy."
+                " Leave empty if using `amount_in`."
+            ),
+        ] = None,
+        min_amount_out: Annotated[
+            Decimal | None,
+            Field(
+                description="The minimum amount of `want` tokens you want to buy."
+                " Leave empty to auto-compute this amount when `amount_in` is provided."
+            ),
+        ] = None,
+        slippage: Annotated[
+            Decimal | None,
+            Field(
+                description="""
+                The maximum change in equilibrium price you are willing to accept.
+                Quantity is a value-less ratio, convert to a ratio if user specifies a percent.
+                Leave empty to use the default of `0.005` (0.5%),
+                or when `max_amount_in`/`min_amount_out` are provided.
+                """
+            ),
+        ] = None,
+    ) -> str:
+        """
+        Performs a token swap, converting an amount of have tokens into want tokens. This function
+        is designed to execute trades on-chain and accounts for real-world dynamics such as
+        slippage and market impact.
+
+        **Important Note**: This function will account for market shifts, which can be set by the
+        `slippage`, `max_amount_in`, or `min_amount_out` parameters. Use these to protect against
+        adverse market changes while executing the user's order.
+        """
+
+        # NOTE: FastMCP doesn't actually support `Decimal` auto-casting yet
+        if isinstance(amount_in, str):
+            amount_in = Decimal(amount_in)
+        if isinstance(max_amount_in, str):
+            max_amount_in = Decimal(max_amount_in)
+        if isinstance(amount_out, str):
+            amount_out = Decimal(amount_out)
+        if isinstance(min_amount_out, str):
+            min_amount_out = Decimal(min_amount_out)
+
+        uni = ctx.request_context.lifespan_context
+        receipt = uni.swap(
+            have=have,
+            want=want,
+            amount_in=amount_in,
+            max_amount_in=max_amount_in,
+            amount_out=amount_out,
+            min_amount_out=min_amount_out,
+            slippage=slippage,
+            sender=account,
+            confirmations_required=0,
+        )
+        return str(receipt.txn_hash)
+
+    server.run()

uniswap_sdk/main.py

@@ -326,36 +326,51 @@ def swap(
         order: Order | None = None,
         routes: Iterable[Route] | None = None,
         receiver: "str | BaseAddress | AddressType | None" = None,
+        native_in: bool = False,
         native_out: bool = False,
         as_transaction: bool = False,
         deadline: timedelta | None = None,
         value: str | int | None = None,
         **order_and_txn_kwargs,
     ) -> "ReceiptAPI | TransactionAPI":
-        order_kwargs: dict = dict()
         if not order:
-            field: str  # NOTE: mypy happy
-            for field in set(ExactInOrder.model_fields) | set(ExactOutOrder.model_fields):
-                if field in order_and_txn_kwargs:
-                    order_kwargs[field] = order_and_txn_kwargs.pop(field)
-
-            if value:
+            order_kwargs: dict = dict()
+            field_name: str  # NOTE: mypy happy
+            for field_name in set(ExactInOrder.model_fields) | set(ExactOutOrder.model_fields):
+                if (
+                    field_name in order_and_txn_kwargs
+                    # NOTE: We want to remove it from `order_and_txn_kwargs` but not use it if None
+                    and (field_value := order_and_txn_kwargs.pop(field_name)) is not None
+                ):
+                    order_kwargs[field_name] = field_value
+
+            if value is not None or native_in or order_kwargs.get("have") in ("ether", "ETH"):
                 order_kwargs["have"] = "WETH"
+                native_in = True
 
-                if "amount_out" in order_kwargs and "max_amount_in" not in order_kwargs:
-                    order_kwargs["max_amount_in"] = self.conversion_manager.convert(value, int)
-
-                elif "amount_in" not in order_kwargs:
-                    order_kwargs["amount_in"] = self.conversion_manager.convert(value, int)
-
-            if native_out or order_kwargs.get("want") == "ether":
+            elif native_out or order_kwargs.get("want") in ("ether", "ETH"):
                 order_kwargs["want"] = "WETH"
                 native_out = True
 
+            if native_in and value is not None:
+                if order_kwargs.get("amount_out") is not None:
+                    if order_kwargs.get("max_amount_in") is None:
+                        order_kwargs["max_amount_in"] = self.conversion_manager.convert(value, int)
+
+                elif order_kwargs.get("amount_in") is None:
+                    order_kwargs["amount_in"] = self.conversion_manager.convert(value, int)
+
             order = self.create_order(**order_kwargs)
 
         permit_step = None
-        if not value:
+        if native_in:
+            if value is None:
+                eth_amount = (
+                    order.amount_in if isinstance(order, ExactInOrder) else order.max_amount_in
+                )
+                value = f"{eth_amount} ether"
+
+        else:
             from ape.api import AccountAPI
 
             if not isinstance(sender := order_and_txn_kwargs.get("sender"), AccountAPI):
@@ -388,10 +403,9 @@ def swap(
             order=order,
             routes=routes,
             permit_step=permit_step,
-            native_in=bool(value),
+            native_in=native_in,
             native_out=native_out,
             receiver=receiver,
-            **order_kwargs,
         )
 
         return (self.router.plan_as_transaction if as_transaction else self.router.execute)(

uniswap_sdk/utils.py

@@ -63,8 +63,10 @@ def get_liquidity(token: TokenInstance, route: Route) -> Decimal:
 
         token = pair.other(token)
 
-    assert liquidity != Decimal("inf")
-    return liquidity
+    if liquidity < Decimal("inf"):
+        return liquidity
+
+    return Decimal(0)
 
 
 def get_total_fee(route: Route) -> Decimal: