Spaces:
Sleeping
Sleeping
File size: 3,187 Bytes
6d8aa95 4facc97 107cced 06ffc6c b57bc26 0ae4701 4facc97 107cced b57bc26 107cced 4facc97 107cced f97da2b 0ae4701 f97da2b 0ae4701 4facc97 0ae4701 f97da2b 4facc97 0ae4701 4facc97 0ae4701 4facc97 107cced 6d8aa95 06ffc6c a28b1b4 53d2fb7 a28b1b4 53d2fb7 b57bc26 4facc97 53d2fb7 4facc97 53d2fb7 4facc97 a28b1b4 4facc97 107cced 4facc97 f97da2b a28b1b4 4facc97 f97da2b 107cced 6d8aa95 107cced 33412db 53d2fb7 33412db |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 |
'''RSS MCP server demonstration client app.'''
import os
import logging
from pathlib import Path
from logging.handlers import RotatingFileHandler
import gradio as gr
import assets.html as html
import client.gradio_functions as gradio_funcs
import client.interface as interface
from client.mcp_client import MCPClientWrapper
from client.anthropic_bridge import AnthropicBridge
# Make sure log directory exists
Path('logs').mkdir(parents=True, exist_ok=True)
# Clear old logs if present
gradio_funcs.delete_old_logs('logs', 'rss_client')
# Set-up logger
logger = logging.getLogger()
logging.basicConfig(
handlers=[RotatingFileHandler(
'logs/rss_client.log',
maxBytes=100000,
backupCount=10,
mode='w'
)],
level=logging.INFO,
format='%(levelname)s - %(name)s - %(message)s'
)
logger = logging.getLogger(__name__)
# Handle MCP server connection and interactions
RSS_CLIENT = MCPClientWrapper(
'https://agents-mcp-hackathon-rss-mcp-server.hf.space/gradio_api/mcp/sse'
)
# Handles Anthropic API I/O
BRIDGE = AnthropicBridge(
RSS_CLIENT,
api_key=os.environ['ANTHROPIC_API_KEY']
)
async def send_message(message: str, chat_history: list) -> str:
'''Submits user message to agent.
Args:
message: the new message from the user as a string
chat_history: list containing conversation history where each element is
a dictionary with keys 'role' and 'content'
Returns
New chat history with model's response to user added.
'''
function_logger = logging.getLogger(__name__ + '.submit_input')
function_logger.info('Submitting user message: %s', message)
chat_history.append({"role": "user", "content": message})
chat_history = await interface.agent_input(BRIDGE, chat_history)
return '', chat_history
with gr.Blocks(title='MCP RSS client') as demo:
with gr.Row():
gr.HTML(html.TITLE)
gr.Markdown(html.DESCRIPTION)
# MCP connection/tool dump
connect_btn = gr.Button('Connect to MCP server')
status = gr.Textbox(label='MCP server tool dump', interactive=False, lines=4)
connect_btn.click(RSS_CLIENT.list_tools, outputs=status) # pylint: disable=no-member
# Log output
logs = gr.Textbox(label='Client logs', lines=10, max_lines=10)
timer = gr.Timer(1, active=True)
timer.tick( # pylint: disable=no-member
lambda: gradio_funcs.update_log(), # pylint: disable=unnecessary-lambda
outputs=logs
)
# Chat interface
chatbot = gr.Chatbot(
value=[],
height=500,
type='messages',
show_copy_button=True
)
msg = gr.Textbox(
'Are there any new posts on Hacker News?',
label='Ask about content or articles on a site or platform',
placeholder='Is there anything new on Hacker News?',
scale=4
)
msg.submit( # pylint: disable=no-member
send_message,
[msg, chatbot],
[msg, chatbot]
)
if __name__ == '__main__':
current_directory = os.getcwd()
if 'pyrite' in current_directory:
demo.launch(server_name='0.0.0.0', server_port=7860)
else:
demo.launch()
|