feat: Python Farcaster Plugin (#221)

* feat: add python farcaster plugin

Co-Authored-By: Agus Armellini Fischer <[email protected]>

* docs: add README and pyproject.toml for farcaster plugin

Co-Authored-By: Agus Armellini Fischer <[email protected]>

---------

Co-authored-by: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
Co-authored-by: Agus Armellini Fischer <[email protected]>

python/src/plugins/farcaster/README.md

@@ -0,0 +1,48 @@
+# Goat Farcaster Plugin 🐐 - Python
+
+Farcaster plugin for Goat. Allows you to create tools for interacting with the Farcaster social protocol through the Neynar API.
+
+## Installation
+```bash
+pip install goat-sdk-plugin-farcaster
+```
+
+## Setup
+
+```python
+from goat_plugins.farcaster import farcaster
+
+plugin = farcaster({ 
+    "api_key": "your_neynar_api_key"
+})
+```
+
+## Features
+
+- Full Farcaster protocol support through Neynar API
+- Cast creation and interaction
+- Thread and conversation management
+- Search functionality
+- Authentication via Signer UUID
+- Proper error handling
+- Python async/await support
+- Type hints with Pydantic models
+
+## API Reference
+
+### Plugin Configuration
+
+| Parameter | Type | Description |
+|-----------|------|-------------|
+| api_key | str | Your Neynar API key |
+| base_url | str | (Optional) Custom API base URL |
+
+## Goat
+
+<div align="center">
+Go out and eat some grass.
+
+[Docs](https://ohmygoat.dev) | [Examples](https://github.com/goat-sdk/goat/tree/main/typescript/examples) | [Discord](https://discord.gg/goat-sdk)</div>
+
+## Goat 🐐
+Goat 🐐 (Great Onchain Agent Toolkit) is an open-source library enabling AI agents to interact with blockchain protocols and smart contracts via their own wallets.

python/src/plugins/farcaster/goat_plugins/farcaster/__init__.py

@@ -0,0 +1,23 @@
+from dataclasses import dataclass
+from typing import Optional
+from goat.classes.plugin_base import PluginBase
+from .service import FarcasterService
+
+
+@dataclass
+class FarcasterPluginOptions:
+    api_key: str
+    base_url: Optional[str] = None
+
+
+class FarcasterPlugin(PluginBase):
+    def __init__(self, options: FarcasterPluginOptions):
+        super().__init__("farcaster", [FarcasterService(options.api_key, options.base_url)])
+
+    def supports_chain(self, chain) -> bool:
+        # farcaster is chain-agnostic
+        return True
+
+
+def farcaster(options: FarcasterPluginOptions) -> FarcasterPlugin:
+    return FarcasterPlugin(options)

python/src/plugins/farcaster/goat_plugins/farcaster/parameters.py

@@ -0,0 +1,26 @@
+from pydantic import BaseModel, Field
+from typing import Optional
+
+
+class GetCastParameters(BaseModel):
+    identifier: str = Field(..., description="Cast URL or hash identifier")
+    type: str = Field("hash", description="Type of identifier (url or hash)")
+
+
+class PublishCastParameters(BaseModel):
+    signer_uuid: str = Field(..., description="Unique ID of the signer publishing the cast")
+    text: str = Field(..., description="Contents of the cast")
+    parent: Optional[str] = Field(None, description="Parent cast hash if this is a reply")
+    channel_id: Optional[str] = Field(None, description="Channel ID if posting to a specific channel")
+
+
+class SearchCastsParameters(BaseModel):
+    query: str = Field(..., description="Text query to find matching casts")
+    limit: Optional[int] = Field(20, description="Max results to retrieve")
+
+
+class GetConversationParameters(BaseModel):
+    identifier: str = Field(..., description="Cast URL or hash identifier")
+    type: str = Field("hash", description="Type of identifier (url or hash)")
+    reply_depth: Optional[int] = Field(2, description="Depth of replies to fetch (0-5)")
+    limit: Optional[int] = Field(20, description="Max results in conversation")

python/src/plugins/farcaster/goat_plugins/farcaster/service.py

@@ -0,0 +1,57 @@
+import aiohttp
+from goat.decorators.tool import Tool
+from .parameters import (
+    GetCastParameters,
+    PublishCastParameters,
+    SearchCastsParameters,
+    GetConversationParameters
+)
+
+
+class FarcasterService:
+    def __init__(self, api_key: str, base_url: str = "https://api.neynar.com/v2/farcaster"):
+        self.api_key = api_key
+        self.base_url = base_url
+
+    @Tool({"description": "Get a cast by its URL or hash", "parameters_schema": GetCastParameters})
+    async def get_cast(self, parameters: dict):
+        url = f"{self.base_url}/cast?identifier={parameters['identifier']}&type={parameters['type']}"
+        return await self._make_request("GET", url)
+
+    @Tool({"description": "Publish a new cast", "parameters_schema": PublishCastParameters})
+    async def publish_cast(self, parameters: dict):
+        url = f"{self.base_url}/cast"
+        return await self._make_request("POST", url, json={
+            "signer_uuid": parameters['signer_uuid'],
+            "text": parameters['text'],
+            "parent": parameters.get('parent'),
+            "channel_id": parameters.get('channel_id'),
+        })
+
+    @Tool({"description": "Search for casts", "parameters_schema": SearchCastsParameters})
+    async def search_casts(self, parameters: dict):
+        url = f"{self.base_url}/cast/search"
+        return await self._make_request("GET", url, params={
+            "q": parameters['query'],
+            "limit": parameters.get('limit', 20)
+        })
+
+    @Tool({"description": "Get a conversation by its URL or hash", "parameters_schema": GetConversationParameters})
+    async def get_conversation(self, parameters: dict):
+        url = f"{self.base_url}/cast/conversation"
+        return await self._make_request("GET", url, params={
+            "identifier": parameters['identifier'],
+            "type": parameters['type'],
+            "reply_depth": parameters.get('reply_depth', 2),
+            "limit": parameters.get('limit', 20),
+        })
+
+    async def _make_request(self, method, url, **kwargs):
+        headers = kwargs.pop("headers", {})
+        headers["x-api-key"] = self.api_key
+        headers["content-type"] = "application/json"
+        async with aiohttp.ClientSession() as session:
+            async with session.request(method, url, headers=headers, **kwargs) as response:
+                if not response.ok:
+                    raise Exception(f"HTTP error! status: {response.status}, text: {await response.text()}")
+                return await response.json()

python/src/plugins/farcaster/pyproject.toml

@@ -0,0 +1,43 @@
+[tool.poetry]
+name = "goat-sdk-plugin-farcaster"
+version = "0.1.1"
+description = "Goat plugin for Farcaster"
+authors = ["Andrea Villa <[email protected]>"]
+readme = "README.md"
+keywords = ["goat", "sdk", "web3", "agents", "ai"]
+homepage = "https://ohmygoat.dev/"
+repository = "https://github.com/goat-sdk/goat"
+packages = [
+    { include = "goat_plugins/farcaster" },
+] 
+
+[tool.poetry.dependencies]
+python = "^3.10"
+aiohttp = "^3.8.6"
+goat-sdk = "^0.1.1"
+
+[tool.poetry.group.test.dependencies]
+pytest = "^8.3.4"
+pytest-asyncio = "^0.25.0"
+
+[tool.poetry.urls]
+"Bug Tracker" = "https://github.com/goat-sdk/goat/issues"
+
+[tool.pytest.ini_options]
+addopts = [
+  "--import-mode=importlib",
+]
+pythonpath = "src"
+asyncio_default_fixture_loop_scope = "function"
+
+[build-system]
+requires = ["poetry-core"]
+build-backend = "poetry.core.masonry.api"
+
+[tool.poetry.group.dev.dependencies]
+ruff = "^0.8.6"
+goat-sdk = { path = "../../goat-sdk", develop = true }
+
+[tool.ruff]
+line-length = 120
+target-version = "py312"