Created
June 18, 2025 16:52
-
-
Save skylarbpayne/f0002905a25adbc29823617827226c5d to your computer and use it in GitHub Desktop.
Reproduction of lilypad error
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
from mirascope import llm, Messages | |
from mirascope.mcp import sse_client | |
import lilypad | |
from pydantic import BaseModel, Field, ValidationError | |
from tenacity import retry, stop_after_attempt | |
from fastmcp import FastMCP | |
from dotenv import load_dotenv | |
import asyncio | |
load_dotenv() | |
lilypad.configure(auto_llm=True) | |
config = { | |
"mcpServers": { | |
"browsermcp": { | |
"command": "npx", | |
"args": ["@browsermcp/mcp@latest"] | |
} | |
} | |
} | |
mcp = FastMCP.as_proxy(config, name="Browser") | |
sse_app = mcp.sse_app() | |
class LinkedInAnalytics(BaseModel): | |
num_followers: int = Field(description="The number of followers you have on LinkedIn") | |
num_followers_growth_pct_7d: float = Field(description="The percentage growth in followers you have on LinkedIn") | |
num_impressions_7d: int = Field(description="The number of impressions you have on LinkedIn") | |
num_profile_viewers_7d: int = Field(description="The number of profile viewers you have on LinkedIn") | |
num_posts_7d: int = Field(description="The number of posts you have on LinkedIn") | |
num_comments_7d: int = Field(description="The number of comments you have on LinkedIn") | |
num_search_appearances_7d: int = Field(description="The number of search appearances you have on LinkedIn") | |
@mcp.tool() | |
async def summarize_linkedin_analytics() -> LinkedInAnalytics: | |
"""Get the LinkedIn analytics.""" | |
res = await run_browser_task(get_linkedin_analytics) | |
return res | |
@lilypad.trace(versioning='automatic') | |
@llm.call(provider="anthropic", model="claude-sonnet-4-20250514", response_model=LinkedInAnalytics) | |
async def get_linkedin_analytics(*, history: list[Messages.Type] | None = None) -> list[Messages.Type]: | |
history = history or [] | |
return [ | |
Messages.System("You are a helpful assistant that can navigate the web. You specialize in getting LinkedIn analytics. Navigate to https://www.linkedin.com/dashboard/ and extract out the relevant metrics."), | |
Messages.User("Please visit linkedin.com and get the analytics for the current page"), | |
*history, | |
] | |
async def run_browser_task_one_step_(task, *args, history: list[Messages.Type], **kwargs): | |
try: | |
result = await task(*args, history=history, **kwargs) | |
except ValidationError as e: | |
print(f'Validation error with {task.__name__}: {e}') | |
history.append(Messages.User(f'Could you try again? It seems like you made a mistake, here is the error: {e}')) | |
return None, history, False | |
if tools := getattr(result, 'tools', None): | |
tools_and_outputs = [] | |
for tool in tools: | |
try: | |
tool_result = await retry(stop=stop_after_attempt(2), before=lambda x: print(f'Trying tool {tool._name()} ({x.attempt_number} / 2)'))(tool.call)() | |
except Exception as e: | |
print(f'Error calling tool {tool._name()}: {e}') | |
tool_result = f'Error calling tool {tool._name()}. Maybe you do not need this tool call: {e}' | |
tools_and_outputs.append((tool, tool_result)) | |
if tools_and_outputs: | |
history.append(result.message_param) | |
history += result.tool_message_params(tools_and_outputs) | |
return result, history, False | |
return result, history, True | |
@lilypad.trace(versioning='automatic') | |
async def run_browser_task(task, *args, **kwargs): | |
async with sse_client("http://127.0.0.1:8082/sse") as client: | |
tools = await client.list_tools() | |
browser_tools = [t for t in tools if t._name().startswith("Browser")] | |
print(f'Found {len(browser_tools)} browser tools') | |
task_fn = llm.override(task, tools=browser_tools) | |
history = [] | |
done = False | |
step = 0 | |
while not done: | |
step += 1 | |
if step > 10: | |
raise Exception('Browser task took too long to complete') | |
result, history, done = await run_browser_task_one_step_(task_fn, *args, history=history, **kwargs) | |
return result | |
if __name__ == "__main__": | |
asyncio.run(run_browser_task(get_linkedin_analytics)) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment