add modelstudio_demos package with 2 demos (#76)

This commit is contained in:
Attan
2026-01-04 17:46:28 +08:00
committed by GitHub
parent f32ef5e059
commit 80421039a0
22 changed files with 2422 additions and 0 deletions

216
modelstudio_demos/chat_demo/.gitignore vendored Normal file
View File

@@ -0,0 +1,216 @@
# Byte-compiled / optimized / DLL files
__pycache__/
*.py[codz]
*$py.class
# C extensions
*.so
# Distribution / packaging
.Python
build/
develop-eggs/
dist/
downloads/
eggs/
.eggs/
lib/
lib64/
parts/
sdist/
var/
wheels/
share/python-wheels/
*.egg-info/
.installed.cfg
*.egg
MANIFEST
# PyInstaller
# Usually these files are written by a python script from a template
# before PyInstaller builds the exe, so as to inject date/other infos into it.
*.manifest
*.spec
# Installer logs
pip-log.txt
pip-delete-this-directory.txt
# Unit test / coverage reports
htmlcov/
.tox/
.nox/
.coverage
.coverage.*
.cache
nosetests.xml
coverage.xml
*.cover
*.py.cover
.hypothesis/
.pytest_cache/
cover/
# Translations
*.mo
*.pot
# Django stuff:
*.log
local_settings.py
db.sqlite3
db.sqlite3-journal
# Flask stuff:
instance/
.webassets-cache
# Scrapy stuff:
.scrapy
# Sphinx documentation
docs/_build/
# PyBuilder
.pybuilder/
target/
# Jupyter Notebook
.ipynb_checkpoints
# IPython
profile_default/
ipython_config.py
# pyenv
# For a library or package, you might want to ignore these files since the code is
# intended to run in multiple environments; otherwise, check them in:
# .python-version
# pipenv
# According to pypa/pipenv#598, it is recommended to include Pipfile.lock in version control.
# However, in case of collaboration, if having platform-specific dependencies or dependencies
# having no cross-platform support, pipenv may install dependencies that don't work, or not
# install all needed dependencies.
# Pipfile.lock
# UV
# Similar to Pipfile.lock, it is generally recommended to include uv.lock in version control.
# This is especially recommended for binary packages to ensure reproducibility, and is more
# commonly ignored for libraries.
# uv.lock
# poetry
# Similar to Pipfile.lock, it is generally recommended to include poetry.lock in version control.
# This is especially recommended for binary packages to ensure reproducibility, and is more
# commonly ignored for libraries.
# https://python-poetry.org/docs/basic-usage/#commit-your-poetrylock-file-to-version-control
# poetry.lock
# poetry.toml
# pdm
# Similar to Pipfile.lock, it is generally recommended to include pdm.lock in version control.
# pdm recommends including project-wide configuration in pdm.toml, but excluding .pdm-python.
# https://pdm-project.org/en/latest/usage/project/#working-with-version-control
# pdm.lock
# pdm.toml
.pdm-python
.pdm-build/
# pixi
# Similar to Pipfile.lock, it is generally recommended to include pixi.lock in version control.
# pixi.lock
# Pixi creates a virtual environment in the .pixi directory, just like venv module creates one
# in the .venv directory. It is recommended not to include this directory in version control.
.pixi
# PEP 582; used by e.g. github.com/David-OConnor/pyflow and github.com/pdm-project/pdm
__pypackages__/
# Celery stuff
celerybeat-schedule
celerybeat.pid
# Redis
*.rdb
*.aof
*.pid
# RabbitMQ
mnesia/
rabbitmq/
rabbitmq-data/
# ActiveMQ
activemq-data/
# SageMath parsed files
*.sage.py
# Environments
.env
.envrc
.venv
env/
venv/
ENV/
env.bak/
venv.bak/
# Spyder project settings
.spyderproject
.spyproject
# Rope project settings
.ropeproject
# mkdocs documentation
/site
# mypy
.mypy_cache/
.dmypy.json
dmypy.json
# Pyre type checker
.pyre/
# pytype static type analyzer
.pytype/
# Cython debug symbols
cython_debug/
# PyCharm
# JetBrains specific template is maintained in a separate JetBrains.gitignore that can
# be found at https://github.com/github/gitignore/blob/main/Global/JetBrains.gitignore
# and can be added to the global gitignore or merged into this file. For a more nuclear
# option (not recommended) you can uncomment the following to ignore the entire idea folder.
# .idea/
# Abstra
# Abstra is an AI-powered process automation framework.
# Ignore directories containing user credentials, local state, and settings.
# Learn more at https://abstra.io/docs
.abstra/
# Visual Studio Code
# Visual Studio Code specific template is maintained in a separate VisualStudioCode.gitignore
# that can be found at https://github.com/github/gitignore/blob/main/Global/VisualStudioCode.gitignore
# and can be added to the global gitignore or merged into this file. However, if you prefer,
# you could uncomment the following to ignore the entire vscode folder
# .vscode/
# Ruff stuff:
.ruff_cache/
# PyPI configuration file
.pypirc
# Marimo
marimo/_static/
marimo/_lsp/
__marimo__/
# Streamlit
.streamlit/secrets.toml

View File

@@ -0,0 +1,45 @@
repos:
# Pre-commit hooks for basic file checks
- repo: https://github.com/pre-commit/pre-commit-hooks
rev: v5.0.0
hooks:
- id: trailing-whitespace
name: Trim trailing whitespace
- id: end-of-file-fixer
name: Fix end of files
- id: check-yaml
name: Check YAML files
- id: check-added-large-files
name: Check for large files
args: ['--maxkb=1000']
- id: check-json
name: Check JSON files
- id: check-toml
name: Check TOML files
- id: mixed-line-ending
name: Check mixed line endings
# Python code formatting with Black
- repo: https://github.com/psf/black
rev: 24.10.0
hooks:
- id: black
name: Format Python code with Black
language_version: python3.10
# Python import sorting with isort
- repo: https://github.com/PyCQA/isort
rev: 5.13.2
hooks:
- id: isort
name: Sort Python imports with isort
args: ["--profile", "black"]
# Python linting with Ruff (fast alternative to flake8)
- repo: https://github.com/astral-sh/ruff-pre-commit
rev: v0.8.4
hooks:
- id: ruff
name: Lint Python code with Ruff
args: [--fix, --exit-non-zero-on-fix]

View File

@@ -0,0 +1,143 @@
# Alibaba Cloud Bailian High-Code Agent Starter
## Project Introduction
This is a starter project based on FastAPI Web framework and AgentScope, providing you with an initial template code package for deploying Agents locally or via Alibaba Cloud Bailian high-code cloud deployment.
It supports direct local running and testing, and deployment to Alibaba Cloud Bailian via uvicorn, allowing you to freely code and assemble atomic capabilities such as LLM, MCP, RAG, memory, and search from Alibaba Cloud Bailian & AgentScope.
## Installing Dependencies
First, make sure you have Python 3.10 or higher installed.
## Local Startup and Testing
```bash
pip install -r requirements.txt
```
### Dependency Description
- `fastapi`: For building Web APIs
- `uvicorn`: For running FastAPI applications
- `agentscope-runtime`: AgentScope runtime environment
- `PyYAML`: PyYAML parsing package
## Configuration
### DashScope API Configuration
To use LLM features, you need to configure the Alibaba Cloud Bailian DashScope API KEY, which can also be added to the deployment machine's environment variables for subsequent cloud deployment:
1. Set `DASHSCOPE_API_KEY` in the `deploy_starter/config.yml` file:
```yaml
DASHSCOPE_API_KEY: "your-api-key-here,sk-xxx"
```
2. Or set it as an environment variable:
```bash
export DASHSCOPE_API_KEY="your-api-key-here,sk-xxx"
```
## Running the Project
### Switch to project root directory and run directly
```bash
cd current-project-root-directory, the directory where setup.py is located
```
```bash
python -m deploy_starter.main
```
### Running with uvicorn
```bash
uvicorn deploy_starter.main:app --host 127.0.0.1 --port 8080 --reload
```
## API Endpoints
### Health Check
Check if the application is running properly:
```bash
curl http://127.0.0.1:8080/health
```
Expected response:
```
"OK"
```
### Chat Endpoint
Chat with the LLM (requires DashScope API key configuration):
```bash
curl -X POST http://127.0.0.1:8080/process \
-H "Content-Type: application/json" \
-d '{"message": "Hello, world!"}'
```
Expected response:
```json
{
"response": "Hello! How can I help you?"
}
```
## Notes
1. Chat functionality will be unavailable if `DASHSCOPE_API_KEY` is not configured.
2. The default model is `qwen-turbo`. You can change `DASHSCOPE_MODEL_NAME` in `config.yml` to switch models.
## Alibaba Cloud Bailian High-Code Cloud Deployment
### Preferably, you can directly upload the code package through the Alibaba Cloud Bailian High-Code Console
[Create Application - High-Code Application](https://bailian.console.aliyun.com//app-center?tab=app#/app-center)
![img_1.png](deploy_by_ui.png)
### Command-line console method for code upload and deployment - Better for quick code modifications and update deployments
#### 1. Install Dependencies
```bash
pip install agentscope-runtime==1.0.0
pip install "agentscope-runtime[deployment]==1.0.0"
```
#### 2. Set Environment Variables
```bash
export ALIBABA_CLOUD_ACCESS_KEY_ID=... # Your Alibaba Cloud account AccessKey (required)
export ALIBABA_CLOUD_ACCESS_KEY_SECRET=... # Your Alibaba Cloud account SecurityKey (required)
# Optional: If you want to use separate OSS AK/SK, you can set the following (if not set, the account AK/SK above will be used). Please ensure the account has OSS read/write permissions.
export MODELSTUDIO_WORKSPACE_ID=... # Your Bailian workspace ID
export OSS_ACCESS_KEY_ID=...
export OSS_ACCESS_KEY_SECRET=...
export OSS_REGION=cn-beijing
```
#### 3. Packaging and Deployment
##### Method A: Manually build wheel file
Ensure your project can be built into a wheel file. You can use setup.py, setup.cfg, or pyproject.toml.
Build the wheel file:
```bash
python setup.py bdist_wheel
```
Deploy:
```bash
runtime-fc-deploy \
--deploy-name [your app name] \
--whl-path [relative path to your wheel file, e.g. "/dist/your_app.whl"]
```
![img.png](deploy_by_cli.png)
For details, please refer to the Alibaba Cloud Bailian High-Code Deployment Documentation: [Alibaba Cloud Bailian High-Code Deployment Documentation](https://bailian.console.aliyun.com/?tab=api#/api/?type=app&url=2983030)

View File

@@ -0,0 +1,143 @@
# 阿里云百炼高代码Agent Starter
## 项目简介
这是一个基于FastAPI Web框架和AgentScope的启动项目能给到你通过本地或者阿里云百炼高代码云端部署Agent的初始模版代码包。
支持直接本地运行测试和通过 uvicorn 部署到阿里云百炼,自由代码编写组装阿里云百炼&AgentScope中的LLM、MCP、RAG、记忆、搜索等原子能力。
## 安装依赖
首先确保你已经安装了 Python 3.10 或更高版本。
## 本地启动测试
```bash
pip install -r requirements.txt
```
### 依赖说明
- `fastapi`: 用于构建 Web API
- `uvicorn`: 用于运行 FastAPI 应用
- `agentscope-runtime`: AgentScope 运行时环境
- `PyYAML`: PyYAML解析包
## 配置
### DashScope API 配置
要使用 LLM 功能,你需要配置阿里云百炼 DashScope API KEY后续云端部署也可以添加到部署机器环境变量中
1.`deploy_starter/config.yml` 文件中设置 `DASHSCOPE_API_KEY`
```yaml
DASHSCOPE_API_KEY: "your-api-key-here,sk-xxx"
```
2. 或者设置环境变量:
```bash
export DASHSCOPE_API_KEY="your-api-key-here,sk-xxx"
```
## 运行项目
### 切换到项目根目录 直接运行
```bash
cd 当前项目根目录,setup.py 文件所在的目录
```
```bash
python -m deploy_starter.main
```
### 使用 uvicorn 运行
```bash
uvicorn deploy_starter.main:app --host 127.0.0.1 --port 8080 --reload
```
## API 接口
### 健康检查
检查应用是否正常运行:
```bash
curl http://127.0.0.1:8080/health
```
预期响应:
```
"OK"
```
### 聊天接口
与 LLM 进行对话(需要配置 DashScope API 密钥):
```bash
curl -X POST http://127.0.0.1:8080/process \
-H "Content-Type: application/json" \
-d '{"message": "你好,世界!"}'
```
预期响应:
```json
{
"response": "你好!有什么我可以帮助你的吗?"
}
```
## 注意事项
1. 如果未配置 `DASHSCOPE_API_KEY`,聊天功能将不可用。
2. 默认使用 `qwen-turbo` 模型,可以在 `config.yml` 中修改 `DASHSCOPE_MODEL_NAME` 来切换模型。
## 阿里云百炼高代码 云端部署
### 优先可以选择阿里云百炼高代码控制台直接上传代码包
[创建应用-高代码应用](https://bailian.console.aliyun.com//app-center?tab=app#/app-center)
![img_1.png](deploy_by_ui.png)
### 命令行console方式进行代码上传部署-更适合快速修改代码进行更新部署
#### 1. 安装依赖
```bash
pip install agentscope-runtime==1.0.0
pip install "agentscope-runtime[deployment]==1.0.0"
```
#### 2. 设置环境变量
```bash
export ALIBABA_CLOUD_ACCESS_KEY_ID=... # 你的阿里云账号AccessKey必填
export ALIBABA_CLOUD_ACCESS_KEY_SECRET=... # 你的阿里云账号SecurityKey必填
# 可选:如果你希望使用单独的 OSS AK/SK可设置如下未设置时将使用到上面的账号 AK/SK请确保账号有 OSS 的读写权限
export MODELSTUDIO_WORKSPACE_ID=... # 你的百炼业务空间id
export OSS_ACCESS_KEY_ID=...
export OSS_ACCESS_KEY_SECRET=...
export OSS_REGION=cn-beijing
```
#### 3. 打包和部署
##### 方式 A手动构建 wheel 文件
确保你的项目可以被构建为 wheel 文件。你可以使用 setup.py、setup.cfg 或 pyproject.toml。
构建 wheel 文件:
```bash
python setup.py bdist_wheel
```
部署:
```bash
runtime-fc-deploy \
--deploy-name [你的应用名称] \
--whl-path [到你的wheel文件的相对路径 如"/dist/your_app.whl"]
```
![img.png](deploy_by_cli.png)
具体请查看阿里云百炼高代码部署文档:[阿里云百炼高代码部署文档](https://bailian.console.aliyun.com/?tab=api#/api/?type=app&url=2983030)

View File

Binary file not shown.

After

Width:  |  Height:  |  Size: 494 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 101 KiB

View File

@@ -0,0 +1,32 @@
# app config
APP_NAME: "ModelStudio-Agent Starter"
DEBUG: false
# Server Config
FC_START_HOST: "0.0.0.0"
HOST: "127.0.0.1"
PORT: 8080
RELOAD: true
# DashScope config
DASHSCOPE_API_KEY:
DASHSCOPE_MODEL_NAME: "qwen-turbo"
# log config
LOG_LEVEL: "INFO"
# Setup config
SETUP_PACKAGE_NAME: "deploy_starter"
SETUP_MODULE_NAME: "main"
SETUP_FUNCTION_NAME: "run_app"
SETUP_COMMAND_NAME: "ModelStudio-Agent-starter"
SETUP_NAME: "ModelStudio-Agent-starter"
SETUP_VERSION: "0.1.0"
SETUP_DESCRIPTION: "ModelStudio-Agent-starter"
SETUP_LONG_DESCRIPTION: "ModelStudio-Agent-starter services, supporting both direct execution and uvicorn deployment"
# observability config
TELEMETRY_ENABLE: TRUE
# FC run config
FC_RUN_CMD: "python3 /code/python/deploy_starter/main.py"

View File

@@ -0,0 +1,158 @@
# -*- coding: utf-8 -*-
import asyncio
import os
from agentscope.agent import ReActAgent
from agentscope.formatter import DashScopeChatFormatter
from agentscope.model import DashScopeChatModel
from agentscope.pipeline import stream_printing_messages
from agentscope.tool import Toolkit, execute_python_code
from agentscope_runtime.adapters.agentscope.memory import (
AgentScopeSessionHistoryMemory,
)
from agentscope_runtime.engine import AgentApp, LocalDeployManager
from agentscope_runtime.engine.schemas.agent_schemas import AgentRequest
from agentscope_runtime.engine.services.agent_state import InMemoryStateService
from agentscope_runtime.engine.services.session_history import (
InMemorySessionHistoryService,
)
from agentscope_runtime.engine.tracing import TraceType, trace
# Read config.yml file
def read_config():
config_path = os.path.join(os.path.dirname(__file__), "config.yml")
config_data = {}
with open(config_path, "r", encoding="utf-8") as config_file:
for line in config_file:
line = line.strip()
if line and not line.startswith("#"):
if ":" in line:
key, value = line.split(":", 1)
key = key.strip()
value = value.strip().strip("\"'")
if value.lower() == "true":
value = True
elif value.lower() == "false":
value = False
elif value.isdigit():
value = int(value)
config_data[key] = value
return config_data
# Read configuration
config = read_config()
agent_app = AgentApp(
app_name=config.get("APP_NAME"),
app_description="A helpful assistant",
)
# Initialize services (required)
@agent_app.init
async def init_func(self):
self.state_service = InMemoryStateService()
self.session_service = InMemorySessionHistoryService()
await self.state_service.start()
await self.session_service.start()
@agent_app.shutdown
async def shutdown_func(self):
await self.state_service.stop()
await self.session_service.stop()
@agent_app.endpoint("/")
def read_root():
return {"hi, i'm running"}
@agent_app.endpoint("/health")
def health_check():
return "OK"
# Default interface implementation for /process
@trace(trace_type=TraceType.LLM, trace_name="llm_func")
@agent_app.query(framework="agentscope")
async def query_func(
self,
msgs,
request: AgentRequest = None,
**kwargs,
):
assert kwargs is not None, "kwargs is Required for query_func"
session_id = request.session_id
user_id = request.user_id
state = await self.state_service.export_state(
session_id=session_id,
user_id=user_id,
)
toolkit = Toolkit()
toolkit.register_tool_function(execute_python_code)
agent = ReActAgent(
name="Friday",
model=DashScopeChatModel(
config.get("DASHSCOPE_MODEL_NAME"),
api_key=os.getenv("DASHSCOPE_API_KEY"),
enable_thinking=True,
stream=True,
),
sys_prompt="You're a helpful assistant named Friday.",
toolkit=toolkit,
memory=AgentScopeSessionHistoryMemory(
service=self.session_service,
session_id=session_id,
user_id=user_id,
),
formatter=DashScopeChatFormatter(),
)
if state:
agent.load_state_dict(state)
async for msg, last in stream_printing_messages(
agents=[agent],
coroutine_task=agent(msgs),
):
yield msg, last
state = agent.state_dict()
await self.state_service.save_state(
user_id=user_id,
session_id=session_id,
state=state,
)
async def main():
"""Deploy AgentScope Runtime using LocalDeployManager"""
deployer = LocalDeployManager(
host=config.get("FC_START_HOST", "127.0.0.1"),
port=config.get("PORT", 8080),
)
# Deploy agent_app
await agent_app.deploy(deployer)
# Keep the service running
print("Service started, press Ctrl+C to stop...")
try:
# Wait indefinitely until interrupted
await asyncio.Event().wait()
except KeyboardInterrupt:
print("\nStopping service...")
if __name__ == "__main__":
try:
asyncio.run(main())
except KeyboardInterrupt:
print("\nService stopped")

View File

@@ -0,0 +1,5 @@
fastapi==0.116.1
uvicorn
agentscope-runtime==1.0.0
setuptools
PyYAML

View File

@@ -0,0 +1,92 @@
# -*- coding: utf-8 -*-
import os
import uuid
from setuptools import setup
# Read dependencies from requirements.txt
with open("requirements.txt", encoding="utf-8") as requirements_file:
requirements = requirements_file.read().splitlines()
# Read config.yml
def read_config():
config_path = os.path.join(
os.path.dirname(__file__),
"deploy_starter",
"config.yml",
)
config_data = {}
with open(config_path, "r", encoding="utf-8") as config_file:
for line in config_file:
line = line.strip()
if line and not line.startswith("#"):
if ":" in line:
key, value = line.split(":", 1)
key = key.strip()
value = value.strip().strip("\"'")
if value.lower() == "true":
value = True
elif value.lower() == "false":
value = False
elif value.isdigit():
value = int(value)
config_data[key] = value
return config_data
# Read README file if exists
def read_readme():
readme_files = ["README.md", "README.rst", "README.txt"]
for filename in readme_files:
if os.path.exists(filename):
with open(filename, "r", encoding="utf-8") as readme_handle:
return readme_handle.read()
return "A FastAPI application with AgentScope runtime"
# Load configuration
config = read_config()
# Extract configuration values
setup_package_name = config.get("SETUP_PACKAGE_NAME", "deploy_starter")
setup_module_name = config.get("SETUP_MODULE_NAME", "main")
setup_function_name = config.get("SETUP_FUNCTION_NAME", "run_app")
setup_command_name = config.get(
"SETUP_COMMAND_NAME",
"ModelStudio-Agent-starter",
)
# Generate package name with UUID suffix
base_name = config.get("SETUP_NAME", "ModelStudio-Agent-starter")
unique_name = f"{base_name}-{uuid.uuid4().hex[:8]}"
# Create package structure
setup(
name=unique_name,
version=config.get("SETUP_VERSION", "0.1.0"),
description=config.get("SETUP_DESCRIPTION", "ModelStudio-Agent-starter"),
long_description=config.get(
"SETUP_LONG_DESCRIPTION",
(
"ModelStudio-Agent-starter services, supporting both direct "
"execution and uvicorn deployment"
),
),
packages=[setup_package_name],
package_dir={setup_package_name: setup_package_name},
install_requires=requirements,
python_requires=">=3.8",
entry_points={
"console_scripts": [
(
f"{setup_command_name}={setup_package_name}."
f"{setup_module_name}:{setup_function_name}"
),
],
},
include_package_data=True,
package_data={
setup_package_name: ["config.yml"],
},
)

Binary file not shown.

After

Width:  |  Height:  |  Size: 1.7 MiB

View File

@@ -0,0 +1,336 @@
# FastMCP Server Development Template
> MCP Server development template based on FastMCP framework, quickly develop and deploy to Alibaba Cloud Bailian high-code platform
## 🎉 Features
Core features of this project:
- **🔧 Modular Architecture**: MCP Server code separated into `mcp_server.py`, main program `main.py` handles routing integration
- **💬 Chat API Integration**: New `/process` endpoint supporting Alibaba Cloud Bailian LLM calls and streaming responses
- **🤖 Intelligent Tool Calling**: LLM can automatically identify and call MCP tools (Function Calling)
- **📡 Unified Service Architecture**: FastAPI + FastMCP integration, one service providing both MCP and Chat functionality
- **🔄 Standardized Responses**: Structured streaming responses based on AgentScope ResponseBuilder
- **🌐 CORS Support**: Cross-origin requests supported for frontend integration
- **🎯 Route Optimization**: MCP Server mounted at `/mcp` path, main app provides more endpoints
## ⚡ Quick Start Locally
### 1. Install Dependencies
```bash
pip install -r requirements.txt
```
### 2. Start Service
```bash
python -m deploy_starter.main
```
### 3. Verify Running
**Health Check:**
```bash
curl http://localhost:8080/health
```
**Test Chat Endpoint:**
```bash
curl -X POST http://localhost:8080/process \
-H "Content-Type: application/json" \
-d '{
"input": [
{
"role": "user",
"content": [{"type": "text", "text": "Hello"}]
}
],
"session_id": "test-session-001",
"stream": true
}'
```
### 4. Recommended: Use MCP Inspector to Verify MCP Server Locally
```bash
npx @modelcontextprotocol/inspector
```
Connect to: `http://localhost:8080/mcp`
![MCP inspector.png](MCP inspector.png)
---
## 🛠️ Develop Your First MCP Tool
Define tools in `deploy_starter/mcp_server.py` using the `@mcp.tool()` decorator:
> **Note**: After refactoring, all MCP tool definitions are in `mcp_server.py`, while `main.py` handles integration and routing
### Example 1: Synchronous Tool (Simple call, average IO performance)
```python
from typing import Annotated
from pydantic import Field
@mcp.tool(
name="add Tool",
description="A simple addition tool example"
)
def add_numbers(
a: Annotated[int, Field(description="add a")],
b: Annotated[int, Field(description="add b")]
) -> int:
return a + b
```
### Example 2: Asynchronous Tool (Async call, high IO performance)
```python
@mcp.tool(
name="Alibaba Cloud Bailian search",
description="Search via Alibaba Cloud Bailian API"
)
async def search_by_modelStudio(
query: Annotated[str, Field(description="Search query statement")],
count: Annotated[int, Field(description="Number of search results returned")] = 5
) -> SearchLiteOutput:
input_data = SearchLiteInput(query=query, count=count)
search_component = ModelstudioSearchLite()
result = await search_component.arun(input_data)
return result
```
**Note**: Async tools require setting the environment variable `DASHSCOPE_API_KEY` to call Bailian services
```bash
export DASHSCOPE_API_KEY='sk-xxxxxx'
```
---
## 📝 Parameter Description Specification
Use `Annotated` + `Field` to add descriptions for each parameter:
```python
from typing import Annotated, Optional
from pydantic import Field
@mcp.tool(
name="your_tool_name", # Tool name (what AI sees)
description="Detailed tool description" # Tool purpose description
)
def your_tool(
param1: Annotated[str, Field(description="Description of parameter 1")],
param2: Annotated[int, Field(description="Description of parameter 2")] = 10
) -> dict:
# Your business logic
return {"result": "success"}
```
---
## Alibaba Cloud Bailian High-Code Cloud Deployment
### Priority option: Upload code package directly through Alibaba Cloud Bailian high-code console
[Create Application - High-Code Application](https://bailian.console.aliyun.com//app-center?tab=app#/app-center)
### Command line console method for code upload and deployment - Better for quick code modifications and update deployments
#### 1. Install Dependencies
```bash
pip install agentscope-runtime==1.0.0
pip install "agentscope-runtime[deployment]==1.0.0"
```
#### 2. Set Environment Variables
```bash
export ALIBABA_CLOUD_ACCESS_KEY_ID=... # Your Alibaba Cloud AccessKey (required)
export ALIBABA_CLOUD_ACCESS_KEY_SECRET=... # Your Alibaba Cloud SecurityKey (required)
# Optional: If you want to use separate OSS AK/SK, you can set the following (if not set, the above account AK/SK will be used), please ensure the account has OSS read/write permissions
export MODELSTUDIO_WORKSPACE_ID=... # Your Bailian workspace ID
export OSS_ACCESS_KEY_ID=...
export OSS_ACCESS_KEY_SECRET=...
export OSS_REGION=cn-beijing
```
#### 3. Package and Deploy
##### Method A: Manually Build Wheel File
Ensure your project can be built as a wheel file. You can use setup.py, setup.cfg, or pyproject.toml.
Build wheel file:
```bash
python setup.py bdist_wheel
```
Deploy:
```bash
runtime-fc-deploy \
--deploy-name [Your application name] \
--whl-path [Relative path to your wheel file e.g. "/dist/your_app.whl"]
```
For details, please refer to the Alibaba Cloud Bailian high-code deployment documentation: [Alibaba Cloud Bailian High-Code Deployment Documentation](https://bailian.console.aliyun.com/?tab=api#/api/?type=app&url=2983030)
---
## 📋 Project Structure
```
.
├── deploy_starter/
│ ├── main.py # Main program - FastAPI app entry, integrates Chat and MCP routing
│ ├── mcp_server.py # MCP Server definition - Define your MCP tools here
│ └── config.yml # Configuration file
├── requirements.txt # Dependency list
├── setup.py # Package configuration (for cloud deployment)
├── README_zh.md # Chinese documentation
└── README_en.md # English documentation
```
**Core Files Description:**
- `main.py`: FastAPI main app, provides `/process` endpoint and lifecycle management, mounts MCP Server at `/mcp` path
- `mcp_server.py`: FastMCP server instance, defines all MCP tools, provides tool list and call functions
---
## 🔧 Configuration
Edit `deploy_starter/config.yml`:
```yaml
# MCP Server Configuration
MCP_SERVER_NAME: "my-mcp-server"
MCP_SERVER_VERSION: "1.0.0"
# Server Configuration
FC_START_HOST: "0.0.0.0" # For cloud deployment
PORT: 8080
HOST: "127.0.0.1" # For local development
# Alibaba Cloud Bailian API Key (optional, can also use environment variable)
# DASHSCOPE_API_KEY: "sk-xxx"
DASHSCOPE_MODEL_NAME: "qwen-plus" # LLM model name
```
### DashScope API Configuration
To use Chat and LLM features, you need to configure the Alibaba Cloud Bailian DashScope API KEY:
1. Set `DASHSCOPE_API_KEY` in `deploy_starter/config.yml`:
```yaml
DASHSCOPE_API_KEY: "sk-xxx"
```
2. Or set it as an environment variable:
```bash
export DASHSCOPE_API_KEY="sk-xxx"
```
---
## 💡 Development Suggestions
### Synchronous vs Asynchronous Tools
- **Synchronous Tools**: Suitable for simple calculations, local operations
```python
@mcp.tool()
def sync_tool(param: str) -> str:
return f"processed: {param}"
```
- **Asynchronous Tools**: Suitable for API calls, database queries, I/O operations
```python
@mcp.tool()
async def async_tool(param: str) -> str:
result = await some_api_call(param)
return result
```
### Tool Naming Conventions
- `name`: Tool name visible to AI (supports Chinese)
- `description`: Detailed explanation of tool purpose, helps AI understand when to call
---
## 🎯 Using in AI Clients
### Claude Desktop
Edit the configuration file `~/Library/Application Support/Claude/claude_desktop_config.json` (macOS):
```json
{
"mcpServers": {
"my-mcp-server": {
"command": "python",
"args": ["-m", "deploy_starter.main"],
"env": {}
}
}
}
```
### Cursor / Cline
Connect to MCP Server URL:
```
http://localhost:8080/mcp
```
### Bailian High-Code Agent Integration
If your application is deployed to Bailian high-code platform, you can directly use the `/process` endpoint for Agent conversations, supporting:
- Natural language interaction
- Automatic tool calling
- Streaming responses
- Complete conversation context management
---
## 📚 API Endpoints
| Endpoint | Method | Description |
|------------|------|------|
| `/` | GET | Server information |
| `/health` | GET | Health check (do not modify) |
| `/process` | POST | Chat endpoint, supports LLM conversation and tool calling (requires DASHSCOPE_API_KEY) |
| `/mcp` | GET/POST | MCP Server endpoint (Streamable HTTP transport) |
### Chat Endpoint Details
**Request Format:**
```json
{
"input": [
{
"role": "user",
"content": [
{"type": "text", "text": "User message"}
]
}
],
"session_id": "Session ID",
"stream": true
}
```
**Response Format:**
- Streaming response (SSE), complies with AgentScope ResponseBuilder standard
- Supports multiple message types: `message` (normal answer), `reasoning` (thinking process), `plugin_call` (tool call), `plugin_call_output` (tool output)
**Core Features:**
- ✅ Automatically identify and call MCP tools
- ✅ Support multi-turn conversation context
- ✅ Streaming response, real-time results
- ✅ Transparent tool calling process

View File

@@ -0,0 +1,337 @@
# FastMCP Server 开发模版
> 基于 FastMCP 框架的 MCP Server 开发模版,快速开发并部署到阿里云百炼高代码
## 🎉 特性
本项目核心功能:
- **🔧 模块化架构**: MCP Server 代码分离至 `mcp_server.py`,主程序 `main.py` 负责路由整合
- **💬 Chat API 集成**: 新增 `/process` 端点,支持阿里云百炼 LLM 调用和流式响应
- **🤖 智能工具调用**: LLM 可自动识别并调用 MCP 工具Function Calling
- **📡 统一服务架构**: FastAPI + FastMCP 集成,一个服务同时提供 MCP 和 Chat 功能
- **🔄 标准化响应**: 基于 AgentScope ResponseBuilder 的结构化流式响应
- **🌐 CORS 支持**: 支持跨域请求,便于前端集成
- **🎯 路由优化**: MCP Server 挂载至 `/mcp` 路径,主应用提供更多端点
## ⚡ 本地快速开始
### 1. 安装依赖
```bash
pip install -r requirements.txt
```
### 2. 启动服务
```bash
python -m deploy_starter.main
```
### 3. 验证运行
**健康检查:**
```bash
curl http://localhost:8080/health
```
**测试 Chat 接口调用天气搜索MCP Tool:**
```bash
curl -X POST http://localhost:8080/process \
-H "Content-Type: application/json" \
-d '{
"input": [
{
"role": "user",
"content": [{"type": "text", "text": "帮我查一下杭州的天气最近5天的"}]
}
],
"session_id": "test-session-001",
"stream": true
}'
```
### 4. 推荐使用MCP Inspector本地先验证MCP server
```bash
npx @modelcontextprotocol/inspector
```
连接地址使用: `http://localhost:8080/mcp`
![MCP inspector.png](MCP inspector.png)
---
## 🛠️ 开发你的第一个 MCP 工具
`deploy_starter/mcp_server.py` 中,使用 `@mcp.tool()` 装饰器定义工具:
> **注意**: 重构后,所有 MCP 工具定义都在 `mcp_server.py` 中,`main.py` 负责集成和路由
### 示例 1: 同步工具简单调用IO性能一般
```python
from typing import Annotated
from pydantic import Field
@mcp.tool(
name="add Tool",
description="一个简单的加法工具示例"
)
def add_numbers(
a: Annotated[int, Field(description="add a")],
b: Annotated[int, Field(description="add b")]
) -> int:
return a + b
```
### 示例 2: 异步工具异步调用IO性能高
```python
@mcp.tool(
name="阿里云百炼search",
description="通过阿里云百炼 API 搜索"
)
async def search_by_modelStudio(
query: Annotated[str, Field(description="搜索的query语句")],
count: Annotated[int, Field(description="搜索返回结果数")] = 5
) -> SearchLiteOutput:
input_data = SearchLiteInput(query=query, count=count)
search_component = ModelstudioSearchLite()
result = await search_component.arun(input_data)
return result
```
**注意**: 异步工具需要设置环境变量 `DASHSCOPE_API_KEY`用来调用百炼服务
```bash
export DASHSCOPE_API_KEY='sk-xxxxxx'
```
---
## 📝 参数描述规范
使用 `Annotated` + `Field` 为每个参数添加描述:
```python
from typing import Annotated, Optional
from pydantic import Field
@mcp.tool(
name="your_tool_name", # 工具名称AI 看到的名字)
description="工具的详细描述" # 工具用途说明
)
def your_tool(
param1: Annotated[str, Field(description="参数1的描述")],
param2: Annotated[int, Field(description="参数2的描述")] = 10
) -> dict:
# 你的业务逻辑
return {"result": "success"}
```
---
## 阿里云百炼高代码 云端部署
### 优先可以选择阿里云百炼高代码控制台直接上传代码包
[创建应用-高代码应用](https://bailian.console.aliyun.com//app-center?tab=app#/app-center)
### 命令行console方式进行代码上传部署-更适合快速修改代码进行更新部署
#### 1. 安装依赖
```bash
pip install agentscope-runtime==1.0.0
pip install "agentscope-runtime[deployment]==1.0.0"
```
#### 2. 设置环境变量
```bash
export ALIBABA_CLOUD_ACCESS_KEY_ID=... # 你的阿里云账号AccessKey必填
export ALIBABA_CLOUD_ACCESS_KEY_SECRET=... # 你的阿里云账号SecurityKey必填
# 可选:如果你希望使用单独的 OSS AK/SK可设置如下未设置时将使用到上面的账号 AK/SK请确保账号有 OSS 的读写权限
export MODELSTUDIO_WORKSPACE_ID=... # 你的百炼业务空间id
export OSS_ACCESS_KEY_ID=...
export OSS_ACCESS_KEY_SECRET=...
export OSS_REGION=cn-beijing
```
#### 3. 打包和部署
##### 方式 A手动构建 wheel 文件
确保你的项目可以被构建为 wheel 文件。你可以使用 setup.py、setup.cfg 或 pyproject.toml。
构建 wheel 文件:
```bash
python setup.py bdist_wheel
```
部署:
```bash
runtime-fc-deploy \
--deploy-name [你的应用名称] \
--whl-path [到你的wheel文件的相对路径 如"/dist/your_app.whl"]
```
具体请查看阿里云百炼高代码部署文档:[阿里云百炼高代码部署文档](https://bailian.console.aliyun.com/?tab=api#/api/?type=app&url=2983030)
---
## 📋 项目结构
```
.
├── deploy_starter/
│ ├── main.py # 主程序 - FastAPI 应用入口,集成 Chat 和 MCP 路由
│ ├── mcp_server.py # MCP Server 定义 - 在这里定义你的 MCP 工具
│ └── config.yml # 配置文件
├── requirements.txt # 依赖列表
├── setup.py # 打包配置(用于云端部署)
├── README_zh.md # 中文文档
└── README_en.md # 英文文档
```
**核心文件说明:**
- `main.py`: FastAPI 主应用,提供 `/process` 端点和生命周期管理,将 MCP Server 挂载到 `/mcp` 路径
- `mcp_server.py`: FastMCP 服务器实例,定义所有 MCP 工具,提供工具列表和调用函数
---
## 🔧 配置说明
编辑 `deploy_starter/config.yml`:
```yaml
# MCP Server 配置
MCP_SERVER_NAME: "my-mcp-server"
MCP_SERVER_VERSION: "1.0.0"
# 服务器配置
FC_START_HOST: "0.0.0.0" # 云端部署使用
PORT: 8080
HOST: "127.0.0.1" # 本地开发使用
# 阿里云百炼 API Key可选也可以用环境变量
# DASHSCOPE_API_KEY: "sk-xxx"
DASHSCOPE_MODEL_NAME: "qwen-plus" # LLM 模型名称
```
### DashScope API 配置
要使用 Chat 和 LLM 功能,需要配置阿里云百炼 DashScope API KEY
1.`deploy_starter/config.yml` 中设置 `DASHSCOPE_API_KEY`:
```yaml
DASHSCOPE_API_KEY: "sk-xxx"
```
2. 或设置为环境变量:
```bash
export DASHSCOPE_API_KEY="sk-xxx"
```
---
## 💡 开发建议
### 同步 vs 异步工具
- **同步工具**: 适合简单计算、本地操作
```python
@mcp.tool()
def sync_tool(param: str) -> str:
return f"processed: {param}"
```
- **异步工具**: 适合 API 调用、数据库查询、I/O 操作
```python
@mcp.tool()
async def async_tool(param: str) -> str:
result = await some_api_call(param)
return result
```
### 工具命名规范
- `name`: AI 看到的工具名称(支持中文)
- `description`: 详细说明工具用途,帮助 AI 理解何时调用
---
## 🎯 在 AI 客户端中使用
### Claude Desktop
编辑配置文件 `~/Library/Application Support/Claude/claude_desktop_config.json` (macOS):
```json
{
"mcpServers": {
"my-mcp-server": {
"command": "python",
"args": ["-m", "deploy_starter.main"],
"env": {}
}
}
}
```
### Cursor / Cline
连接 MCP Server URL:
```
http://localhost:8080/mcp
```
### 百炼高代码 Agent 集成
如果你的应用部署到百炼高代码,可以直接使用 `/process` 端点进行 Agent 对话,支持:
- 自然语言交互
- 自动工具调用
- 流式响应
- 完整的对话上下文管理
---
## 📚 API 端点
| 端点 | 方法 | 说明 |
|------------|------|------|
| `/` | GET | 服务器信息 |
| `/health` | GET | 健康检查(请勿修改) |
| `/process` | POST | Chat 接口,支持 LLM 对话和工具调用(需要 DASHSCOPE_API_KEY |
| `/mcp` | GET/POST | MCP Server 端点Streamable HTTP 传输) |
### Chat 接口详细说明
**请求格式:**
```json
{
"input": [
{
"role": "user",
"content": [
{"type": "text", "text": "用户消息"}
]
}
],
"session_id": "会话ID",
"stream": true
}
```
**响应格式:**
- 流式响应SSE符合 AgentScope ResponseBuilder 标准
- 支持多种消息类型: `message`(普通回答)、`reasoning`(思考过程)、`plugin_call`(工具调用)、`plugin_call_output`(工具输出)
**核心特性:**
- ✅ 自动识别并调用 MCP 工具
- ✅ 支持多轮对话上下文
- ✅ 流式响应,实时返回结果
- ✅ 工具调用过程透明可见

View File

@@ -0,0 +1,7 @@
# -*- coding: utf-8 -*-
"""
FastMCP Server Development Template
A starter template for developing MCP servers with FastAPI and FastMCP.
"""
__version__ = "0.1.0"

View File

@@ -0,0 +1,36 @@
# ==================== Application Configuration ====================
APP_NAME: "MCP Server Starter"
DEBUG: false
# ==================== Server Configuration ====================
# Use 0.0.0.0 for cloud deployment; 127.0.0.1 for local development
FC_START_HOST: "0.0.0.0"
HOST: "127.0.0.1"
PORT: 8080
RELOAD: true
# ==================== MCP Configuration ====================
# MCP Server settings
MCP_SERVER_NAME: "my-mcp-server"
MCP_SERVER_VERSION: "1.0.0"
# ==================== Logging Configuration ====================
LOG_LEVEL: "INFO"
# ==================== Packaging Configuration ====================
# Settings for build and distribution
SETUP_PACKAGE_NAME: "deploy_starter"
SETUP_MODULE_NAME: "main"
SETUP_FUNCTION_NAME: "run_app"
SETUP_COMMAND_NAME: "mcp-server-starter"
SETUP_NAME: "mcp-server-starter"
SETUP_VERSION: "0.1.0"
SETUP_DESCRIPTION: "FastMCP Server Starter - template for fast MCP Server development"
SETUP_LONG_DESCRIPTION: "MCP Server starter based on FastMCP framework, supports local run and cloud deployment"
# ==================== Observability Configuration ====================
TELEMETRY_ENABLE: TRUE
# ==================== Cloud Deployment Configuration ====================
# Alibaba Cloud Function Compute (FC) deployment settings
FC_RUN_CMD: "python3 /code/python/deploy_starter/main.py"

View File

@@ -0,0 +1,478 @@
# -*- coding: utf-8 -*-
# flake8: noqa
# pylint: disable=line-too-long,too-many-branches,too-many-statements,too-many-nested-blocks
"""
FastMCP Server Development Template
This is an MCP Server starter template based on the fastMcp framework, allowing developers to quickly develop their own MCP Server and deploy it to Alibaba Cloud Bailian high-code platform
Core features:
1. Use @mcp.tool() decorator to quickly define tools
2. Built-in health check interface
3. Support for HTTP SSE, streamable connection methods
4. Provide complete MCP protocol support (list tools, call tool, etc.)
Developers only need to focus on writing their own tool functions.
"""
import json
import os
from contextlib import asynccontextmanager
from typing import Any
import uvicorn
from agentscope_runtime.engine.helpers.agent_api_builder import ResponseBuilder
from agentscope_runtime.engine.schemas.agent_schemas import Role
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from openai import AsyncOpenAI
from pydantic import BaseModel
from starlette.middleware.cors import CORSMiddleware
# Import MCP Server instance
from deploy_starter.mcp_server import (
call_mcp_tool,
convert_mcp_tools_to_openai_format,
list_mcp_tools,
mcp,
)
# ==================== Configuration Reading ====================
def read_config():
"""Read config.yml file"""
config_path = os.path.join(os.path.dirname(__file__), "config.yml")
config_data = {}
with open(config_path, encoding="utf-8") as config_file:
for line in config_file:
line = line.strip()
if line and not line.startswith("#"):
if ":" in line:
key, value = line.split(":", 1)
key = key.strip()
value = value.strip().strip("\"'")
if value.lower() == "true":
value = True
elif value.lower() == "false":
value = False
elif value.isdigit():
value = int(value)
config_data[key] = value
return config_data
config = read_config()
# ==================== Create MCP ASGI Application ====================
# Pre-create MCP application instance for reuse in lifespan and mount
mcp_asgi_app = mcp.streamable_http_app(path="/")
@asynccontextmanager
async def lifespan(fastapi_app: FastAPI):
"""Application lifecycle management - integrate MCP application's lifespan"""
# Use MCP application's lifespan context manager
async with mcp_asgi_app.router.lifespan_context(fastapi_app):
# Application startup completed, entering running state
yield
# Automatic cleanup when application closes
# Create FastAPI application
app = FastAPI(
title=config.get("APP_NAME", "MCP Server with Chat"),
debug=config.get("DEBUG", False),
lifespan=lifespan,
)
app.add_middleware(
CORSMiddleware,
allow_origins=["http://localhost:5173"], # or ["*"] for development only
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# ==================== Mount MCP Server Routes ====================
# Integrate MCP Server routes into main application
# This way only one server needs to be started to provide both MCP tools and Chat interface
# Note: mcp_asgi_app is already created above, use it directly here
# Mount MCP routes to main application under /mcp path
app.mount("/mcp", mcp_asgi_app)
@app.get("/")
def read_root():
return "<h1>hi, i'm running</h1>"
@app.get("/health")
def health_check():
return "OK"
class ContentItem(BaseModel):
type: str # e.g.: "text", "data", etc.
text: str | None = None # Text content (optional)
data: dict[str, Any] | None = None # Data content (optional)
status: str | None = None # Status
class Config:
extra = "allow" # Allow extra fields
class MessageItem(BaseModel):
role: str # e.g.: "user", "assistant"
content: list[ContentItem] | None = None # content array (optional)
type: str | None = (
None # Message type: message, plugin_call, plugin_call_output, etc.
)
class Config:
extra = "allow" # Allow extra fields (like sequence_number, object, status, id, etc.)
class ChatRequest(BaseModel):
input: list[MessageItem] # Message array
session_id: str # Session ID
stream: bool | None = True # Whether to stream response
# ==================== Chat Interface Implementation ====================
# Bailian example process call interface, requires DASHSCOPE_API_KEY configuration
@app.post("/process")
async def chat(request_data: ChatRequest):
"""
Chat interface implementation, supports LLM calls and MCP tool calls
Core workflow:
1. Receive user message
2. Get MCP tool list
3. Call LLM (with function calling)
4. If LLM needs to call tools, call MCP tools
5. Return tool results to LLM
6. Return final response (conforms to AgentScope ResponseBuilder format)
"""
# Get DashScope API Key
api_key = os.environ.get("DASHSCOPE_API_KEY")
if not api_key:
api_key = config.get("DASHSCOPE_API_KEY")
if not api_key:
return {"error": "DASHSCOPE_API_KEY not configured"}
# Initialize OpenAI client (DashScope is compatible with OpenAI API)
client = AsyncOpenAI(
api_key=api_key,
base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
)
# Convert message format to OpenAI format
# Keep conversation history: user messages + assistant's final answer (type="message")
# Ignore intermediate steps: plugin_call, plugin_call_output, reasoning
messages = []
for msg in request_data.input:
# Process user messages
if msg.role == "user":
content_text = ""
if msg.content:
for content_item in msg.content:
if content_item.type == "text" and content_item.text:
content_text += content_item.text
if content_text: # Only add non-empty messages
messages.append({"role": "user", "content": content_text})
# Process assistant's final answer (type="message")
elif msg.role == "assistant" and msg.type == "message":
content_text = ""
if msg.content:
for content_item in msg.content:
if content_item.type == "text" and content_item.text:
content_text += content_item.text
if content_text:
messages.append({"role": "assistant", "content": content_text})
# Get MCP tool list
try:
mcp_tools = await list_mcp_tools()
openai_tools = convert_mcp_tools_to_openai_format(mcp_tools)
except Exception as e:
print(f"Failed to get MCP tools: {e}")
openai_tools = []
async def generate_response():
"""Generate streaming response - conforms to Bailian Response/Message/Content architecture"""
# Create ResponseBuilder
response_builder = ResponseBuilder(
session_id=request_data.session_id,
response_id=f"resp_{request_data.session_id}",
)
# 1. Send Response created status
yield f"data: {response_builder.created().model_dump_json()}\n\n"
# 2. Send Response in_progress status
yield f"data: {response_builder.in_progress().model_dump_json()}\n\n"
try:
# First phase: LLM initial response (may contain tool call decisions)
if openai_tools:
response = await client.chat.completions.create(
model=config.get("DASHSCOPE_MODEL_NAME", "qwen-plus"),
messages=messages,
tools=openai_tools,
stream=True,
)
else:
response = await client.chat.completions.create(
model=config.get("DASHSCOPE_MODEL_NAME", "qwen-plus"),
messages=messages,
stream=True,
)
# Collect LLM response content and tool calls
llm_content = ""
tool_calls = []
current_tool_call: dict[str, Any] | None = None
async for chunk in response:
if chunk.choices and len(chunk.choices) > 0:
choice = chunk.choices[0]
delta = choice.delta
# Collect text content
if delta.content:
llm_content += delta.content
# Collect tool calls
if delta.tool_calls:
for tool_call_chunk in delta.tool_calls:
if tool_call_chunk.index is not None:
if current_tool_call is None:
pass
elif (
current_tool_call["index"]
!= tool_call_chunk.index
):
tool_calls.append(current_tool_call)
current_tool_call = None
if current_tool_call is None:
current_tool_call = {
"index": tool_call_chunk.index,
"id": tool_call_chunk.id or "",
"type": "function",
"function": {
"name": tool_call_chunk.function.name
or "",
"arguments": (
tool_call_chunk.function.arguments
or ""
),
},
}
elif tool_call_chunk.function.arguments:
current_tool_call["function"][
"arguments"
] += tool_call_chunk.function.arguments
if current_tool_call:
tool_calls.append(current_tool_call)
# Decide message flow based on whether there are tool calls
if tool_calls:
# Scenario: has tool calls
# 3. Create reasoning message (if LLM has thinking content)
if llm_content.strip():
reasoning_msg_builder = (
response_builder.create_message_builder(
role=Role.ASSISTANT,
message_type="reasoning",
)
)
yield f"data: {reasoning_msg_builder.get_message_data().model_dump_json()}\n\n"
reasoning_content_builder = (
reasoning_msg_builder.create_content_builder()
)
yield f"data: {reasoning_content_builder.add_text_delta(llm_content).model_dump_json()}\n\n"
yield f"data: {reasoning_content_builder.complete().model_dump_json()}\n\n"
yield f"data: {reasoning_msg_builder.complete().model_dump_json()}\n\n"
# 4. First add assistant message (containing all tool calls) to message history
messages.append(
{
"role": "assistant",
"content": None,
"tool_calls": tool_calls,
},
)
# 5. Process each tool call
for tool_call in tool_calls:
tool_name = tool_call["function"]["name"]
tool_args = json.loads(tool_call["function"]["arguments"])
# 5.1 Create plugin_call message (display to user)
plugin_call_msg_builder = (
response_builder.create_message_builder(
role=Role.ASSISTANT,
message_type="plugin_call",
)
)
yield f"data: {plugin_call_msg_builder.get_message_data().model_dump_json()}\n\n"
plugin_call_content_builder = (
plugin_call_msg_builder.create_content_builder(
content_type="data",
)
)
tool_call_data = {
"name": tool_name,
"arguments": json.dumps(tool_args, ensure_ascii=False),
}
yield f"data: {plugin_call_content_builder.add_data_delta(tool_call_data).model_dump_json()}\n\n"
yield f"data: {plugin_call_content_builder.complete().model_dump_json()}\n\n"
yield f"data: {plugin_call_msg_builder.complete().model_dump_json()}\n\n"
# 5.2 Call MCP tool
try:
tool_result = await call_mcp_tool(tool_name, tool_args)
# 5.3 Create plugin_call_output message (display to user)
plugin_output_msg_builder = (
response_builder.create_message_builder(
role=Role.ASSISTANT,
message_type="plugin_call_output",
)
)
yield f"data: {plugin_output_msg_builder.get_message_data().model_dump_json()}\n\n"
plugin_output_content_builder = (
plugin_output_msg_builder.create_content_builder(
content_type="data",
)
)
output_data = {
"name": tool_name,
"output": (
json.dumps(tool_result, ensure_ascii=False)
if tool_result
else ""
),
}
yield f"data: {plugin_output_content_builder.add_data_delta(output_data).model_dump_json()}\n\n"
yield f"data: {plugin_output_content_builder.complete().model_dump_json()}\n\n"
yield f"data: {plugin_output_msg_builder.complete().model_dump_json()}\n\n"
# Add tool message to message history
messages.append(
{
"role": "tool",
"tool_call_id": tool_call["id"],
"content": (
json.dumps(tool_result, ensure_ascii=False)
if tool_result
else ""
),
},
)
except Exception as e:
print(f"Tool call failed: {e}")
# Add error result to message history
messages.append(
{
"role": "tool",
"tool_call_id": tool_call["id"],
"content": f"Error: {str(e)}",
},
)
# 6. Use tool results to call LLM again to generate final answer
final_response = await client.chat.completions.create(
model=config.get("DASHSCOPE_MODEL_NAME", "qwen-plus"),
messages=messages,
stream=True,
)
# 7. Create final message (answer based on tool results)
final_msg_builder = response_builder.create_message_builder(
role=Role.ASSISTANT,
message_type="message",
)
yield f"data: {final_msg_builder.get_message_data().model_dump_json()}\n\n"
final_content_builder = (
final_msg_builder.create_content_builder()
)
async for chunk in final_response:
if chunk.choices and len(chunk.choices) > 0:
choice = chunk.choices[0]
if choice.delta.content:
yield f"data: {final_content_builder.add_text_delta(choice.delta.content).model_dump_json()}\n\n"
yield f"data: {final_content_builder.complete().model_dump_json()}\n\n"
yield f"data: {final_msg_builder.complete().model_dump_json()}\n\n"
else:
# Scenario: no tool calls, return LLM response directly
# 3. Create message (direct answer)
msg_builder = response_builder.create_message_builder(
role=Role.ASSISTANT,
message_type="message",
)
yield f"data: {msg_builder.get_message_data().model_dump_json()}\n\n"
content_builder = msg_builder.create_content_builder()
yield f"data: {content_builder.add_text_delta(llm_content).model_dump_json()}\n\n"
yield f"data: {content_builder.complete().model_dump_json()}\n\n"
yield f"data: {msg_builder.complete().model_dump_json()}\n\n"
# 8. Complete Response
yield f"data: {response_builder.completed().model_dump_json()}\n\n"
# yield "data: [DONE]\n\n"
except Exception as e:
# Error handling
print(f"Chat interface error: {e}")
error_msg_builder = response_builder.create_message_builder(
role=Role.ASSISTANT,
message_type="error",
)
error_content_builder = error_msg_builder.create_content_builder()
error_text = f"Error occurred: {str(e)}"
yield f"data: {error_content_builder.add_text_delta(error_text).model_dump_json()}\n\n"
yield f"data: {error_content_builder.complete().model_dump_json()}\n\n"
yield f"data: {error_msg_builder.complete().model_dump_json()}\n\n"
yield f"data: {response_builder.completed().model_dump_json()}\n\n"
# yield "data: [DONE]\n\n"
return StreamingResponse(
generate_response(),
media_type="text/event-stream",
)
# ==================== Start Application ====================
def run_app():
"""Entry point for running the application via command line."""
uvicorn.run(
"deploy_starter.main:app",
host=config.get("FC_START_HOST", "127.0.0.1"),
port=config.get("PORT", 8080),
reload=config.get("RELOAD", False),
)
if __name__ == "__main__":
# Start FastAPI application
run_app()

View File

@@ -0,0 +1,226 @@
# -*- coding: utf-8 -*-
# flake8: noqa
# pylint: disable=line-too-long
"""
FastMCP Server Development Template
This is an MCP Server starter template based on the fastMcp framework, allowing developers to quickly develop their own MCP Server and deploy it to Alibaba Cloud Bailian high-code platform
Core features:
1. Use @mcp.tool() decorator to quickly define tools
2. Built-in health check interface
3. Support for HTTP SSE, streamable connection methods
4. Provide complete MCP protocol support (list tools, call tool, etc.)
Developers only need to focus on writing their own tool functions.
"""
import json
import os
from typing import Annotated, Any
from agentscope_runtime.tools import ModelstudioSearchLite
from agentscope_runtime.tools.searches import SearchLiteInput, SearchLiteOutput
from fastmcp import Client, FastMCP
from pydantic import Field
# ==================== Configuration Reading ====================
def read_config():
"""Read config.yml file"""
config_path = os.path.join(os.path.dirname(__file__), "config.yml")
config_data = {}
with open(config_path, encoding="utf-8") as config_file:
for line in config_file:
line = line.strip()
if line and not line.startswith("#"):
if ":" in line:
key, value = line.split(":", 1)
key = key.strip()
value = value.strip().strip("\"'")
if value.lower() == "true":
value = True
elif value.lower() == "false":
value = False
elif value.isdigit():
value = int(value)
config_data[key] = value
return config_data
config = read_config()
# ==================== Initialize FastMCP ====================
# Create MCP server instance, define MCP name and version
mcp = FastMCP(
name=config.get("MCP_SERVER_NAME", "my-mcp-server"),
version="1.0.0",
)
# ==================== Tool Definition Examples ====================
# Developers can define their own tools here, using @mcp.tool() decorator
# Example tool1, simple addition tool, simple call with average IO performance
@mcp.tool(
name="add Tool", # Custom tool name for the LLM
description="A simple addition tool example for calculating the sum of two integers", # Custom description
)
def add_numbers(
a: Annotated[int, Field(description="add a")],
b: Annotated[int, Field(description="add b")],
) -> int:
return a + b
# Example tool2, Alibaba Cloud Bailian search, asynchronous call with high IO performance
@mcp.tool(
name="Alibaba Cloud Bailian search", # Custom tool name for the LLM
description="Search MCP wrapper by calling Alibaba Cloud Bailian search API, requires dashScope api key in environment variables", # Custom description
)
async def search_by_modelStudio(
query: Annotated[str, Field(description="Search query statement")],
count: Annotated[
int,
Field(description="Number of search results returned"),
] = 5,
) -> SearchLiteOutput:
input_data = SearchLiteInput(query=query, count=count)
search_component = ModelstudioSearchLite()
result = await search_component.arun(input_data)
print(result)
return result
# ==================== MCP Tool Call Helper Functions ====================
# Use FastMCP Client standard API for tool listing and calling
async def list_mcp_tools() -> list[dict[str, Any]]:
"""
Get MCP tool list using FastMCP Client via StreamableHttpTransport
Connect to MCP Server via HTTP URL, using standard Streamable HTTP transport protocol.
This approach is more suitable for production environments and easier to debug and monitor.
"""
mcp_base_url = (
f"http://{config.get('HOST', '127.0.0.1')}:{config.get('PORT', 8080)}"
)
print(f"\n{'=' * 60}")
print("📋 [MCP Call] Get tool list")
print(f"{'=' * 60}")
print(f"Connection URL: {mcp_base_url}/mcp/")
print("Transport method: StreamableHttpTransport")
try:
# Create FastMCP Client, pass HTTP URL
# Client will automatically infer to use HTTP transport
client = Client(f"{mcp_base_url}/mcp/")
async with client:
# Use standard list_tools() method
tools = await client.list_tools()
# Convert to dictionary format for subsequent processing
tools_list = []
for tool in tools:
tool_dict = {
"name": tool.name,
"description": tool.description or "",
"inputSchema": tool.inputSchema,
}
tools_list.append(tool_dict)
print(f"✅ Successfully retrieved {len(tools_list)} tools")
for i, tool in enumerate(tools_list, 1):
print(f" {i}. {tool['name']} - {tool['description']}")
print(f"{'=' * 60}\n")
return tools_list
except Exception as e:
print(f"❌ Failed to get tool list: {e}")
print(f"{'=' * 60}\n")
return []
async def call_mcp_tool(tool_name: str, arguments: dict[str, Any]) -> Any:
"""
Call MCP tool using FastMCP Client via StreamableHttpTransport
Connect to MCP Server via HTTP URL, using standard Streamable HTTP transport protocol.
This approach is more suitable for production environments and easier to debug and monitor.
"""
mcp_base_url = (
f"http://{config.get('HOST', '127.0.0.1')}:{config.get('PORT', 8080)}"
)
print(f"\n{'=' * 60}")
print("🔧 [MCP Call] Execute tool")
print(f"{'=' * 60}")
print(f"Connection URL: {mcp_base_url}/mcp/")
print("Transport method: StreamableHttpTransport")
print(f"Tool name: {tool_name}")
print(
f"Tool arguments: {json.dumps(arguments, indent=2, ensure_ascii=False)}",
)
try:
# Create FastMCP Client, pass HTTP URL
# Client will automatically infer to use HTTP transport
client = Client(f"{mcp_base_url}/mcp/")
async with client:
# Use standard call_tool() method
result = await client.call_tool(tool_name, arguments)
# Process result
# result.content is a list containing the content returned by the tool
result_data = None
if result.content:
# Extract text content
for content_item in result.content:
if hasattr(content_item, "text"):
result_data = content_item.text
break
if hasattr(content_item, "data"):
result_data = content_item.data
break
print("✅ Tool execution successful")
print(f"Result: {result_data}")
print(f"{'=' * 60}\n")
return result_data
except Exception as e:
print(f"❌ Tool execution failed: {e}")
print(f"{'=' * 60}\n")
return None
def convert_mcp_tools_to_openai_format(
mcp_tools: list[dict[str, Any]],
) -> list[dict[str, Any]]:
"""
Convert MCP tool format to OpenAI function calling format
"""
openai_tools = []
for tool in mcp_tools:
openai_tool = {
"type": "function",
"function": {
"name": tool.get("name", ""),
"description": tool.get("description", ""),
"parameters": tool.get(
"inputSchema",
{"type": "object", "properties": {}, "required": []},
),
},
}
openai_tools.append(openai_tool)
return openai_tools

View File

@@ -0,0 +1,67 @@
[build-system]
requires = ["setuptools>=45", "wheel"]
build-backend = "setuptools.build_meta"
[tool.black]
line-length = 88
target-version = ['py310']
include = '\.pyi?$'
extend-exclude = '''
/(
# directories
\.eggs
| \.git
| \.hg
| \.mypy_cache
| \.tox
| \.venv
| venv
| build
| dist
)/
'''
[tool.isort]
profile = "black"
line_length = 88
skip_gitignore = true
skip = ["venv", "build", "dist", ".eggs"]
known_first_party = ["deploy_starter"]
[tool.ruff]
line-length = 88
target-version = "py310"
extend-exclude = [
"venv",
"build",
"dist",
".eggs",
"*.egg-info",
]
[tool.ruff.lint]
select = [
"E", # pycodestyle errors
"W", # pycodestyle warnings
"F", # pyflakes
"I", # isort
"B", # flake8-bugbear
"C4", # flake8-comprehensions
"UP", # pyupgrade
]
ignore = [
"E501", # line too long (handled by black)
"B008", # do not perform function calls in argument defaults
"C901", # too complex
]
[tool.ruff.lint.per-file-ignores]
"__init__.py" = ["F401"] # Allow unused imports in __init__.py
[tool.bandit]
exclude_dirs = ["tests", "venv", "build", "dist"]
skips = ["B101", "B601"] # Skip assert_used and shell_injection for subprocess
[tool.bandit.assert_used]
skips = ["*_test.py", "test_*.py"]

View File

@@ -0,0 +1,12 @@
fastapi==0.116.1
uvicorn
fastmcp==2.12.5
agentscope-runtime==1.0.0
PyYAML
psutil==7.1.2
shortuuid==1.0.13
steel-sdk==0.13.0
docker==7.1.0
kubernetes==34.1.0
httpx
openai

View File

@@ -0,0 +1,89 @@
# -*- coding: utf-8 -*-
import os
import uuid
from setuptools import setup
# Read dependencies from requirements.txt
with open("requirements.txt", encoding="utf-8") as requirements_file:
requirements = requirements_file.read().splitlines()
# Read config.yml file
def read_config():
config_path = os.path.join(
os.path.dirname(__file__),
"deploy_starter",
"config.yml",
)
config_data = {}
with open(config_path, encoding="utf-8") as config_file:
for line in config_file:
line = line.strip()
if line and not line.startswith("#"):
if ":" in line:
key, value = line.split(":", 1)
key = key.strip()
value = value.strip().strip("\"'")
if value.lower() == "true":
value = True
elif value.lower() == "false":
value = False
elif value.isdigit():
value = int(value)
config_data[key] = value
return config_data
# Read README.md file
def read_readme():
readme_files = ["README.md", "README.rst", "README.txt"]
for filename in readme_files:
if os.path.exists(filename):
with open(filename, encoding="utf-8") as readme_handle:
return readme_handle.read()
return "A FastAPI application with AgentScope runtime"
# Read configuration
config = read_config()
# Get configuration values
setup_package_name = config.get("SETUP_PACKAGE_NAME", "deploy_starter")
setup_module_name = config.get("SETUP_MODULE_NAME", "main")
setup_function_name = config.get("SETUP_FUNCTION_NAME", "run_app")
setup_command_name = config.get("SETUP_COMMAND_NAME", "MCP-Server-starter")
# Generate package name with UUID
base_name = config.get("SETUP_NAME", "MCP-Server-starter")
unique_name = f"{base_name}-{uuid.uuid4().hex[:8]}"
# Create package structure
setup(
name=unique_name,
version=config.get("SETUP_VERSION", "0.1.0"),
description=config.get("SETUP_DESCRIPTION", "MCP-Server-starter"),
long_description=config.get(
"SETUP_LONG_DESCRIPTION",
(
"MCP-Server-starter services, supporting both direct execution "
"and uvicorn deployment"
),
),
packages=[setup_package_name],
package_dir={setup_package_name: setup_package_name},
install_requires=requirements,
python_requires=">=3.8",
entry_points={
"console_scripts": [
(
f"{setup_command_name}={setup_package_name}."
f"{setup_module_name}:{setup_function_name}"
),
],
},
include_package_data=True,
package_data={
setup_package_name: ["config.yml"],
},
)