github/ChatDev
1
0
Fork 0
mirror of https://github.com/OpenBMB/ChatDev.git synced 2026-04-25 11:18:06 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #584 from NA-Wen/main

Browse Source 
Enable OpenClaw Integration with ChatDev Backend Workflow Support
This commit is contained in:
Shu Yao 2026-03-17 21:07:48 +08:00 committed by GitHub
commit 34ecbe5b1c
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
17 changed files with 1077 additions and 729 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
Makefile
View File

@ -10,7 +10,7 @@ dev: ## Run both backend and frontend development servers
.PHONY: server
server: ## Start the backend server in the background
@echo "Starting server in background..."
@uv run python server_main.py --port 6400 --reload &
@uv run python server_main.py --port 6400 &
.PHONY: client
client: ## Start the frontend development server

27
README-zh.md
View File

@ -170,6 +170,33 @@ make dev
```
检查所有 YAML 文件的语法与 schema 错误。
### 🦞 使用 OpenClaw 运行
OpenClaw 可以与 ChatDev 集成，通过 **调用已有的 agent 团队**，或在 ChatDev 中 **动态创建新的 agent 团队** 来完成任务。
开始使用：
1. 启动 ChatDev 2.0 后端。
2. 为你的 OpenClaw 实例安装所需的技能：
```bash
clawdhub install chatdev
```
3. 让 OpenClaw 创建一个 ChatDev 工作流。例如：
* **自动化信息收集与内容发布**
```
创建一个 ChatDev 工作流，用于自动收集热点信息，生成一篇小红书文案，并发布该内容
```
* **多智能体地缘政治模拟**
```
创建一个 ChatDev 工作流，构建多个 agent，用于模拟中东局势未来可能的发展
```
### 🐳 使用 Docker 运行
你也可以通过 Docker Compose 运行整个应用。该方式可简化依赖管理，并提供一致的运行环境。

23
README.md
View File

@ -182,6 +182,29 @@ make dev
```
Checks all YAML files for syntax and schema errors.
### 🦞 Run with OpenClaw
OpenClaw can integrate with ChatDev by invoking existing agent teams or dynamically creating new agent teams within ChatDev.
To get started:
1. Start the ChatDev 2.0 backend.
2. Install the required skills for your OpenClaw instance:
```bash
clawdhub install chatdev
```
3. Ask your OpenClaw to create a ChatDev workflow. For example:
* **Automated information collection and content publishing**
```
Create a ChatDev workflow to automatically collect trending information, generate a Xiaohongshu post, and publish it.
```
* **Multi-agent geopolitical simulation**
```
Create a ChatDev workflow with multiple agents to simulate possible future developments of the Middle East situation.
```
### 🐳 Run with Docker
Alternatively, you can run the entire application using Docker Compose. This method simplifies dependency management and provides a consistent environment.

70
entity/configs/node/tooling.py
View File

@ -312,6 +312,8 @@ class McpRemoteConfig(BaseConfig):
server: str
headers: Dict[str, str] = field(default_factory=dict)
timeout: float | None = None
cache_ttl: float = 0.0
tool_sources: List[str] | None = None
FIELD_SPECS = {
"server": ConfigFieldSpec(
@ -337,6 +339,22 @@ class McpRemoteConfig(BaseConfig):
description="Per-request timeout in seconds",
advance=True,
),
"cache_ttl": ConfigFieldSpec(
name="cache_ttl",
display_name="Tool Cache TTL",
type_hint="float",
required=False,
description="Seconds to cache MCP tool list; 0 disables cache for hot updates",
advance=True,
),
"tool_sources": ConfigFieldSpec(
name="tool_sources",
display_name="Tool Sources Filter",
type_hint="list[str]",
required=False,
description="Only include MCP tools whose meta.source is in this list; omit to default to ['mcp_tools'].",
advance=True,
),
}
@classmethod
@ -360,7 +378,40 @@ class McpRemoteConfig(BaseConfig):
else:
raise ConfigError("timeout must be numeric", extend_path(path, "timeout"))
return cls(server=server, headers=headers, timeout=timeout, path=path)
cache_ttl_value = mapping.get("cache_ttl", 0.0)
if cache_ttl_value is None:
cache_ttl = 0.0
elif isinstance(cache_ttl_value, (int, float)):
cache_ttl = float(cache_ttl_value)
else:
raise ConfigError("cache_ttl must be numeric", extend_path(path, "cache_ttl"))
tool_sources_raw = mapping.get("tool_sources")
tool_sources: List[str] | None = None
if tool_sources_raw is not None:
entries = ensure_list(tool_sources_raw)
normalized: List[str] = []
for idx, entry in enumerate(entries):
if not isinstance(entry, str):
raise ConfigError(
"tool_sources must be a list of strings",
extend_path(path, f"tool_sources[{idx}]"),
)
value = entry.strip()
if value:
normalized.append(value)
tool_sources = normalized
else:
tool_sources = ["mcp_tools"]
return cls(
server=server,
headers=headers,
timeout=timeout,
cache_ttl=cache_ttl,
tool_sources=tool_sources,
path=path,
)
def cache_key(self) -> str:
payload = (
@ -380,6 +431,7 @@ class McpLocalConfig(BaseConfig):
inherit_env: bool = True
startup_timeout: float = 10.0
wait_for_log: str | None = None
cache_ttl: float = 0.0
FIELD_SPECS = {
"command": ConfigFieldSpec(
@ -438,6 +490,14 @@ class McpLocalConfig(BaseConfig):
description="Regex that marks readiness when matched against stdout",
advance=True,
),
"cache_ttl": ConfigFieldSpec(
name="cache_ttl",
display_name="Tool Cache TTL",
type_hint="float",
required=False,
description="Seconds to cache MCP tool list; 0 disables cache for hot updates",
advance=True,
),
}
@classmethod
@ -474,6 +534,13 @@ class McpLocalConfig(BaseConfig):
raise ConfigError("startup_timeout must be numeric", extend_path(path, "startup_timeout"))
wait_for_log = optional_str(mapping, "wait_for_log", path)
cache_ttl_value = mapping.get("cache_ttl", 0.0)
if cache_ttl_value is None:
cache_ttl = 0.0
elif isinstance(cache_ttl_value, (int, float)):
cache_ttl = float(cache_ttl_value)
else:
raise ConfigError("cache_ttl must be numeric", extend_path(path, "cache_ttl"))
return cls(
command=command,
args=normalized_args,
@ -482,6 +549,7 @@ class McpLocalConfig(BaseConfig):
inherit_env=bool(inherit_env),
startup_timeout=startup_timeout,
wait_for_log=wait_for_log,
cache_ttl=cache_ttl,
path=path,
)

14
frontend/package-lock.json generated
View File

@ -1114,7 +1114,8 @@
"version": "5.0.0",
"resolved": "https://registry.npmjs.org/@types/linkify-it/-/linkify-it-5.0.0.tgz",
"integrity": "sha512-sVDA58zAw4eWAffKOaQH5/5j3XeayukzDk+ewSsnv3p4yJEZHCCzMDiZM8e0OUrRvmpGZ85jf4yDHkHsgBNr9Q==",
"license": "MIT"
"license": "MIT",
"peer": true
},
"node_modules/@types/markdown-it": {
"version": "14.1.2",
@ -1131,7 +1132,8 @@
"version": "2.0.0",
"resolved": "https://registry.npmjs.org/@types/mdurl/-/mdurl-2.0.0.tgz",
"integrity": "sha512-RGdgjQUZba5p6QEFAVx2OGb8rQDL/cPRG7GiedRzMcJ1tYnUANBncjbSB1NRGwbvjcPeikRABz2nshyPk1bhWg==",
"license": "MIT"
"license": "MIT",
"peer": true
},
"node_modules/@types/web-bluetooth": {
"version": "0.0.20",
@ -1181,7 +1183,6 @@
"resolved": "https://registry.npmjs.org/@vue-flow/core/-/core-1.48.1.tgz",
"integrity": "sha512-3IxaMBLvWRbznZ4CuK0kVhp4Y4lCDQx9nhi48Swp6PwPw29KNhmiKd2kaBogYeWjGLb/tLjlE9V0s3jEmKCYWw==",
"license": "MIT",
"peer": true,
"dependencies": {
"@vueuse/core": "^10.5.0",
"d3-drag": "^3.0.0",
@ -1419,7 +1420,6 @@
"integrity": "sha512-NZyJarBfL7nWwIq+FDL6Zp/yHEhePMNnnJ0y3qfieCrmNvYct8uvtiV41UvlSe6apAfk0fY1FbWx+NwfmpvtTg==",
"dev": true,
"license": "MIT",
"peer": true,
"bin": {
"acorn": "bin/acorn"
},
@ -1664,7 +1664,6 @@
"resolved": "https://registry.npmjs.org/d3-selection/-/d3-selection-3.0.0.tgz",
"integrity": "sha512-fmTRWbNMmsmWq6xJV8D19U/gw/bwrHfNXxrIN+HfZgnzqTHp9jOmKMhsTUjXOJnZOdZY9Q28y4yebKzqDKlxlQ==",
"license": "ISC",
"peer": true,
"engines": {
"node": ">=12"
}
@ -1811,7 +1810,6 @@
"integrity": "sha512-LEyamqS7W5HB3ujJyvi0HQK/dtVINZvd5mAAp9eT5S/ujByGjiZLCzPcHVzuXbpJDJF/cxwHlfceVUDZ2lnSTw==",
"dev": true,
"license": "MIT",
"peer": true,
"dependencies": {
"@eslint-community/eslint-utils": "^4.8.0",
"@eslint-community/regexpp": "^4.12.1",
@ -2309,7 +2307,6 @@
"resolved": "https://registry.npmjs.org/markdown-it/-/markdown-it-14.1.0.tgz",
"integrity": "sha512-a54IwgWPaeBCAAsv13YgmALOF1elABB08FxO9i+r4VFk5Vl4pKokRPeX8u5TCgSsPi6ec1otfLjdOpVcgbpshg==",
"license": "MIT",
"peer": true,
"dependencies": {
"argparse": "^2.0.1",
"entities": "^4.4.0",
@ -2491,7 +2488,6 @@
"integrity": "sha512-5gTmgEY/sqK6gFXLIsQNH19lWb4ebPDLA4SdLP7dsWkIXHWlG66oPuVvXSGFPppYZz8ZDZq0dYYrbHfBCVUb1Q==",
"dev": true,
"license": "MIT",
"peer": true,
"engines": {
"node": ">=12"
},
@ -2755,7 +2751,6 @@
"integrity": "sha512-dZwN5L1VlUBewiP6H9s2+B3e3Jg96D0vzN+Ry73sOefebhYr9f94wwkMNN/9ouoU8pV1BqA1d1zGk8928cx0rg==",
"dev": true,
"license": "MIT",
"peer": true,
"dependencies": {
"esbuild": "^0.27.0",
"fdir": "^6.5.0",
@ -2830,7 +2825,6 @@
"resolved": "https://registry.npmjs.org/vue/-/vue-3.5.26.tgz",
"integrity": "sha512-SJ/NTccVyAoNUJmkM9KUqPcYlY+u8OVL1X5EW9RIs3ch5H2uERxyyIUI4MRxVCSOiEcupX9xNGde1tL9ZKpimA==",
"license": "MIT",
"peer": true,
"dependencies": {
"@vue/compiler-dom": "3.5.26",
"@vue/compiler-sfc": "3.5.26",

51
frontend/src/pages/LaunchView.vue
View File

@ -70,7 +70,7 @@
>
<CollapsibleMessage
v-if="message.text"
:html-content="renderMarkdown(message.text)"
:html-content="message.htmlContent || renderMarkdown(message.text)"
:raw-content="message.text"
:default-expanded="configStore.AUTO_EXPAND_MESSAGES"
/>
@ -556,6 +556,7 @@ const addTotalLoadingMessage = (nodeId) => {
type: 'dialogue',
name: nodeId,
text: '',
htmlContent: '',
avatar,
isRight: false,
isLoading: true,
@ -597,6 +598,10 @@ const addLoadingEntry = (nodeId, baseKey, label) => {
nodeState.entryMap.set(key, entry)
nodeState.baseKeyToKey.set(baseKey, key)
nodeState.message.loadingEntries.push(entry)
runningLoadingEntries.value += 1
if (runningLoadingEntries.value === 1) {
startLoadingTimer()
}
return entry
}
@ -609,27 +614,56 @@ const finishLoadingEntry = (nodeId, baseKey) => {
const entry = key ? nodeState.entryMap.get(key) : null
if (!entry) return null
const wasRunning = entry.status === 'running'
entry.status = 'done'
entry.endedAt = Date.now()
nodeState.baseKeyToKey.delete(baseKey)
if (wasRunning) {
runningLoadingEntries.value = Math.max(0, runningLoadingEntries.value - 1)
if (runningLoadingEntries.value === 0) {
stopLoadingTimer()
}
}
return entry
}
// Finish all running entries when a node ends or cancels
const finalizeAllLoadingEntries = (nodeState, endedAt = Date.now()) => {
if (!nodeState) return
let finishedCount = 0
for (const entry of nodeState.entryMap.values()) {
if (entry.status === 'running') {
entry.status = 'done'
entry.endedAt = endedAt
finishedCount += 1
}
}
nodeState.baseKeyToKey.clear()
if (finishedCount) {
runningLoadingEntries.value = Math.max(0, runningLoadingEntries.value - finishedCount)
if (runningLoadingEntries.value === 0) {
stopLoadingTimer()
}
}
}
// Global timer for updating loading bubble durations
const now = ref(Date.now())
let loadingTimerInterval = null
const runningLoadingEntries = ref(0)
const startLoadingTimer = () => {
if (loadingTimerInterval) return
loadingTimerInterval = setInterval(() => {
now.value = Date.now()
}, 1000)
}
const stopLoadingTimer = () => {
if (!loadingTimerInterval) return
clearInterval(loadingTimerInterval)
loadingTimerInterval = null
}
// Map sprites for different roles
const nameToSpriteMap = ref(new Map())
@ -878,10 +912,12 @@ const addDialogue = (name, message) => {
const isRight = name === "User"
const htmlContent = renderMarkdown(text)
chatMessages.value.push({
type: 'dialogue',
name: name,
text: text,
htmlContent,
avatar: avatar,
isRight: isRight,
timestamp: Date.now()
@ -1499,13 +1535,6 @@ onMounted(() => {
document.addEventListener('click', handleClickOutside)
document.addEventListener('keydown', handleKeydown)
loadWorkflows()
// Start the global timer
if (!loadingTimerInterval) {
loadingTimerInterval = setInterval(() => {
now.value = Date.now()
}, 1000)
}
})
onUnmounted(() => {
@ -1515,10 +1544,8 @@ onUnmounted(() => {
resetConnectionState()
cleanupRecording()
if (loadingTimerInterval) {
clearInterval(loadingTimerInterval)
loadingTimerInterval = null
}
stopLoadingTimer()
runningLoadingEntries.value = 0
})
const { fromObject, fitView, onPaneReady, onNodesInitialized, setNodes, setEdges, edges } = useVueFlow()

8
frontend/src/utils/apiFunctions.js
View File

@ -69,7 +69,7 @@ export async function postYaml(filename, content) {
export async function updateYaml(filename, content) {
try {
const yamlFilename = addYamlSuffix(filename)
const response = await fetch(apiUrl(`/api/workflows/${encodeURIComponent(yamlFilename)}`), {
const response = await fetch(apiUrl(`/api/workflows/${encodeURIComponent(yamlFilename)}/update`), {
method: 'PUT',
headers: {
'Content-Type': 'application/json',
@ -197,7 +197,7 @@ export async function fetchWorkflowsWithDesc() {
const filesWithDesc = await Promise.all(
data.workflows.map(async (filename) => {
try {
const response = await fetch(apiUrl(`/api/workflows/${encodeURIComponent(filename)}`))
const response = await fetch(apiUrl(`/api/workflows/${encodeURIComponent(filename)}/desc`))
const fileData = await response.json()
return {
name: filename,
@ -234,7 +234,7 @@ export async function fetchWorkflowsWithDesc() {
// Fetch YAML file content
export async function fetchWorkflowYAML(filename) {
try {
const response = await fetch(apiUrl(`/api/workflows/${encodeURIComponent(filename)}`))
const response = await fetch(apiUrl(`/api/workflows/${encodeURIComponent(filename)}/get`))
if (!response.ok) {
throw new Error(`Failed to load YAML file: ${filename}, status: ${response.status}`)
}
@ -250,7 +250,7 @@ export async function fetchWorkflowYAML(filename) {
export async function fetchYaml(filename) {
try {
const yamlFilename = addYamlSuffix(filename)
const response = await fetch(apiUrl(`/api/workflows/${encodeURIComponent(yamlFilename)}`))
const response = await fetch(apiUrl(`/api/workflows/${encodeURIComponent(yamlFilename)}/get`))
const data = await response.json().catch(() => ({}))

4
server/app.py
View File

@ -1,6 +1,8 @@
from fastapi import FastAPI
from server.bootstrap import init_app
from utils.env_loader import load_dotenv_file
load_dotenv_file()
app = FastAPI(title="DevAll Workflow Server", version="1.0.0")
init_app(app)

9
server/bootstrap.py
View File

@ -1,6 +1,7 @@
"""Application bootstrap helpers for the FastAPI server."""
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from server import state
from server.config_schema_router import router as config_schema_router
@ -12,6 +13,14 @@ from utils.middleware import add_middleware
def init_app(app: FastAPI) -> None:
"""Apply shared middleware, routers, and global state to ``app``."""
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
add_exception_handlers(app)
add_middleware(app)

11
server/models.py
View File

@ -1,6 +1,6 @@
"""Pydantic models shared across server routes."""
from typing import List, Literal, Optional
from typing import Any, Dict, List, Literal, Optional
from pydantic import BaseModel, constr
@ -13,6 +13,15 @@ class WorkflowRequest(BaseModel):
log_level: Literal["INFO", "DEBUG"] = "INFO"
class WorkflowRunRequest(BaseModel):
yaml_file: str
task_prompt: str
attachments: Optional[List[str]] = None
session_name: Optional[str] = None
variables: Optional[Dict[str, Any]] = None
log_level: Optional[Literal["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"]] = None
class WorkflowUploadContentRequest(BaseModel):
filename: str
content: str

6
server/routes/__init__.py
View File

@ -1,6 +1,6 @@
"""Aggregates API routers."""
from . import artifacts, batch, execute, health, sessions, uploads, vuegraphs, workflows, websocket
from . import artifacts, execute, execute_sync, health, sessions, uploads, vuegraphs, workflows, websocket, batch, tools
ALL_ROUTERS = [
health.router,
@ -11,7 +11,9 @@ ALL_ROUTERS = [
sessions.router,
batch.router,
execute.router,
execute_sync.router,
tools.router,
websocket.router,
]
__all__ = ["ALL_ROUTERS"]
__all__ = ["ALL_ROUTERS"]

253
server/routes/execute_sync.py Normal file
View File

@ -0,0 +1,253 @@
from __future__ import annotations
import json
import queue
import threading
from datetime import datetime
from pathlib import Path
from typing import Any, Optional, Sequence, Union
from fastapi import APIRouter, HTTPException, Request
from fastapi.responses import StreamingResponse
from starlette.concurrency import run_in_threadpool
from check.check import load_config
from entity.enums import LogLevel
from entity.graph_config import GraphConfig
from entity.messages import Message
from runtime.bootstrap.schema import ensure_schema_registry_populated
from runtime.sdk import OUTPUT_ROOT, run_workflow
from server.models import WorkflowRunRequest
from server.settings import YAML_DIR
from utils.attachments import AttachmentStore
from utils.exceptions import ValidationError, WorkflowExecutionError
from utils.logger import WorkflowLogger
from utils.structured_logger import get_server_logger, LogType
from utils.task_input import TaskInputBuilder
from workflow.graph import GraphExecutor
from workflow.graph_context import GraphContext
router = APIRouter()
_SSE_CONTENT_TYPE = "text/event-stream"
def _normalize_session_name(yaml_path: Path, session_name: Optional[str]) -> str:
if session_name and session_name.strip():
return session_name.strip()
timestamp = datetime.now().strftime("%Y%m%d%H%M%S")
return f"sdk_{yaml_path.stem}_{timestamp}"
def _resolve_yaml_path(yaml_file: Union[str, Path]) -> Path:
candidate = Path(yaml_file).expanduser()
if candidate.is_absolute():
return candidate
if candidate.exists():
return candidate
repo_root = Path(__file__).resolve().parents[2]
yaml_root = YAML_DIR if YAML_DIR.is_absolute() else (repo_root / YAML_DIR)
return (yaml_root / candidate).expanduser()
def _build_task_input(
graph_context: GraphContext,
prompt: str,
attachments: Sequence[Union[str, Path]],
) -> Union[str, list[Message]]:
if not attachments:
return prompt
attachments_dir = graph_context.directory / "code_workspace" / "attachments"
attachments_dir.mkdir(parents=True, exist_ok=True)
store = AttachmentStore(attachments_dir)
builder = TaskInputBuilder(store)
normalized_paths = [str(Path(path).expanduser()) for path in attachments]
return builder.build_from_file_paths(prompt, normalized_paths)
def _run_workflow_with_logger(
*,
yaml_file: Union[str, Path],
task_prompt: str,
attachments: Optional[Sequence[Union[str, Path]]],
session_name: Optional[str],
variables: Optional[dict],
log_level: Optional[LogLevel],
log_callback,
) -> tuple[Optional[Message], dict[str, Any]]:
ensure_schema_registry_populated()
yaml_path = _resolve_yaml_path(yaml_file)
if not yaml_path.exists():
raise FileNotFoundError(f"YAML file not found: {yaml_path}")
attachments = attachments or []
if (not task_prompt or not task_prompt.strip()) and not attachments:
raise ValidationError(
"Task prompt cannot be empty",
details={"task_prompt_provided": bool(task_prompt)},
)
design = load_config(yaml_path, vars_override=variables)
normalized_session = _normalize_session_name(yaml_path, session_name)
graph_config = GraphConfig.from_definition(
design.graph,
name=normalized_session,
output_root=OUTPUT_ROOT,
source_path=str(yaml_path),
vars=design.vars,
)
if log_level:
graph_config.log_level = log_level
graph_config.definition.log_level = log_level
graph_context = GraphContext(config=graph_config)
task_input = _build_task_input(graph_context, task_prompt, attachments)
class _StreamingWorkflowLogger(WorkflowLogger):
def add_log(self, *args, **kwargs):
entry = super().add_log(*args, **kwargs)
if entry:
payload = entry.to_dict()
payload.pop("details", None)
log_callback("log", payload)
return entry
class _StreamingExecutor(GraphExecutor):
def _create_logger(self) -> WorkflowLogger:
level = log_level or self.graph.log_level
return _StreamingWorkflowLogger(
self.graph.name,
level,
use_structured_logging=True,
log_to_console=False,
)
executor = _StreamingExecutor(graph_context, session_id=normalized_session)
executor._execute(task_input)
final_message = executor.get_final_output_message()
logger = executor.log_manager.get_logger() if executor.log_manager else None
log_id = logger.workflow_id if logger else None
token_usage = executor.token_tracker.get_token_usage() if executor.token_tracker else None
meta = {
"session_name": normalized_session,
"yaml_file": str(yaml_path),
"log_id": log_id,
"token_usage": token_usage,
"output_dir": graph_context.directory,
}
return final_message, meta
def _sse_event(event_type: str, data: Any) -> str:
payload = json.dumps(data, ensure_ascii=False, default=str)
return f"event: {event_type}\ndata: {payload}\n\n"
@router.post("/api/workflow/run")
async def run_workflow_sync(request: WorkflowRunRequest, http_request: Request):
try:
resolved_log_level: Optional[LogLevel] = None
if request.log_level:
resolved_log_level = LogLevel(request.log_level)
except ValueError:
raise HTTPException(
status_code=400,
detail="log_level must be one of DEBUG, INFO, WARNING, ERROR, CRITICAL",
)
accepts_stream = _SSE_CONTENT_TYPE in (http_request.headers.get("accept") or "")
if not accepts_stream:
try:
result = await run_in_threadpool(
run_workflow,
request.yaml_file,
task_prompt=request.task_prompt,
attachments=request.attachments,
session_name=request.session_name,
variables=request.variables,
log_level=resolved_log_level,
)
except FileNotFoundError as exc:
raise HTTPException(status_code=404, detail=str(exc))
except ValidationError as exc:
raise HTTPException(status_code=400, detail=str(exc))
except Exception as exc:
logger = get_server_logger()
logger.log_exception(exc, "Failed to run workflow via sync API")
raise WorkflowExecutionError(f"Failed to run workflow: {exc}")
final_message = result.final_message.text_content() if result.final_message else ""
meta = result.meta_info
logger = get_server_logger()
logger.info(
"Workflow execution completed via sync API",
log_type=LogType.WORKFLOW,
session_id=meta.session_name,
yaml_path=meta.yaml_file,
)
return {
"status": "completed",
"final_message": final_message,
"token_usage": meta.token_usage,
"output_dir": str(meta.output_dir.resolve()),
}
event_queue: queue.Queue[tuple[str, Any]] = queue.Queue()
done_event = threading.Event()
def enqueue(event_type: str, data: Any) -> None:
event_queue.put((event_type, data))
def worker() -> None:
try:
enqueue(
"started",
{"yaml_file": request.yaml_file, "task_prompt": request.task_prompt},
)
final_message, meta = _run_workflow_with_logger(
yaml_file=request.yaml_file,
task_prompt=request.task_prompt,
attachments=request.attachments,
session_name=request.session_name,
variables=request.variables,
log_level=resolved_log_level,
log_callback=enqueue,
)
enqueue(
"completed",
{
"status": "completed",
"final_message": final_message.text_content() if final_message else "",
"token_usage": meta["token_usage"],
"output_dir": str(meta["output_dir"].resolve()),
},
)
except (FileNotFoundError, ValidationError) as exc:
enqueue("error", {"message": str(exc)})
except Exception as exc:
logger = get_server_logger()
logger.log_exception(exc, "Failed to run workflow via streaming API")
enqueue("error", {"message": f"Failed to run workflow: {exc}"})
finally:
done_event.set()
threading.Thread(target=worker, daemon=True).start()
async def stream():
while True:
try:
event_type, data = event_queue.get(timeout=0.1)
yield _sse_event(event_type, data)
except queue.Empty:
if done_event.is_set():
break
return StreamingResponse(stream(), media_type=_SSE_CONTENT_TYPE)

74
server/routes/tools.py Normal file
View File

@ -0,0 +1,74 @@
import re
from pathlib import Path
from fastapi import APIRouter, HTTPException
from pydantic import BaseModel, constr
from utils.function_catalog import get_function_catalog
from utils.function_manager import FUNCTION_CALLING_DIR
router = APIRouter()
class LocalToolCreateRequest(BaseModel):
filename: constr(strip_whitespace=True, min_length=1, max_length=255)
content: str
overwrite: bool = False
@router.get("/api/tools/local")
def list_local_tools():
catalog = get_function_catalog()
metadata = catalog.list_metadata()
tools = []
for name, meta in metadata.items():
tools.append(
{
"name": name,
"description": meta.description,
"parameters": meta.parameters_schema,
"module": meta.module_name,
"file_path": meta.file_path,
}
)
tools.sort(key=lambda item: item["name"])
return {
"success": True,
"count": len(tools),
"tools": tools,
"load_error": str(catalog.load_error) if catalog.load_error else None,
}
@router.post("/api/tools/local")
def create_local_tool(payload: LocalToolCreateRequest):
raw_name = payload.filename.strip()
if not raw_name:
raise HTTPException(status_code=400, detail="filename is required")
if not re.match(r"^[A-Za-z0-9_-]+(\.py)?$", raw_name):
raise HTTPException(status_code=400, detail="filename must be alphanumeric with optional .py extension")
filename = raw_name if raw_name.endswith(".py") else f"{raw_name}.py"
tools_dir = Path(FUNCTION_CALLING_DIR).resolve()
tools_dir.mkdir(parents=True, exist_ok=True)
target_path = (tools_dir / filename).resolve()
try:
target_path.relative_to(tools_dir)
except ValueError:
raise HTTPException(status_code=400, detail="filename resolves outside function tools directory")
if target_path.exists() and not payload.overwrite:
raise HTTPException(status_code=409, detail="tool file already exists")
target_path.write_text(payload.content, encoding="utf-8")
catalog = get_function_catalog()
catalog.refresh()
return {
"success": True,
"filename": filename,
"path": str(target_path),
"load_error": str(catalog.load_error) if catalog.load_error else None,
}

156
server/routes/workflows.py
View File

@ -1,4 +1,5 @@
from fastapi import APIRouter, HTTPException
from typing import Any
from server.models import (
WorkflowCopyRequest,
@ -67,6 +68,154 @@ async def list_workflows():
return {"workflows": [file.name for file in YAML_DIR.glob("*.yaml")]}
@router.get("/api/workflows/{filename}/args")
async def get_workflow_args(filename: str):
print(str)
try:
safe_filename = validate_workflow_filename(filename, require_yaml_extension=True)
print(safe_filename)
file_path = YAML_DIR / safe_filename
if not file_path.exists() or not file_path.is_file():
raise ResourceNotFoundError(
"Workflow file not found",
resource_type="workflow",
resource_id=safe_filename,
)
# Load and validate YAML content
raw_content = file_path.read_text(encoding="utf-8")
_, yaml_content = validate_workflow_content(safe_filename, raw_content)
args: list[dict[str, Any]] = []
if isinstance(yaml_content, dict):
graph = yaml_content.get("graph") or {}
if isinstance(graph, dict):
raw_args = graph.get("args") or []
if isinstance(raw_args, list):
if len(raw_args) == 0:
raise ResourceNotFoundError(
"Workflow file does not have args",
resource_type="workflow",
resource_id=safe_filename,
)
for item in raw_args:
# Each item is expected to be like: { arg_name: [ {key: value}, ... ] }
if not isinstance(item, dict) or len(item) != 1:
continue
(arg_name, spec_list), = item.items()
if not isinstance(arg_name, str):
continue
arg_info: dict[str, Any] = {"name": arg_name}
if isinstance(spec_list, list):
for spec in spec_list:
if isinstance(spec, dict):
for key, value in spec.items():
# Later entries override earlier ones if duplicated
arg_info[str(key)] = value
args.append(arg_info)
logger = get_server_logger()
logger.info(
"Workflow args retrieved",
log_type=LogType.WORKFLOW,
filename=safe_filename,
args_count=len(args),
)
return {"args": args}
except ValidationError as exc:
# 参数或文件名等校验错误
raise HTTPException(
status_code=400,
detail={"message": str(exc)},
)
except SecurityError as exc:
# 安全相关错误（例如路径遍历）
raise HTTPException(
status_code=400,
detail={"message": str(exc)},
)
except ResourceNotFoundError as exc:
# 文件不存在
raise HTTPException(
status_code=404,
detail={"message": str(exc)},
)
except Exception as exc:
logger = get_server_logger()
logger.log_exception(exc, f"Unexpected error retrieving workflow args: {filename}")
# 兜底错误
raise HTTPException(
status_code=500,
detail={"message": f"Failed to retrieve workflow args: {exc}"},
)
@router.get("/api/workflows/{filename}/desc")
async def get_workflow_desc(filename: str):
try:
safe_filename = validate_workflow_filename(filename, require_yaml_extension=True)
file_path = YAML_DIR / safe_filename
if not file_path.exists() or not file_path.is_file():
raise ResourceNotFoundError(
"Workflow file not found",
resource_type="workflow",
resource_id=safe_filename,
)
# Load and validate YAML content
raw_content = file_path.read_text(encoding="utf-8")
_, yaml_content = validate_workflow_content(safe_filename, raw_content)
desc = ""
if isinstance(yaml_content, dict):
graph = yaml_content.get("graph") or {}
if isinstance(graph, dict):
desc = graph.get("description") or ""
if len(desc) == 0:
raise ResourceNotFoundError(
"Workflow file does not have args",
resource_type="workflow",
resource_id=safe_filename,
)
logger = get_server_logger()
logger.info(
"Workflow description retrieved",
log_type=LogType.WORKFLOW,
filename=safe_filename,
)
return {"description": desc}
except ValidationError as exc:
# 参数或文件名等校验错误
raise HTTPException(
status_code=400,
detail={"message": str(exc)},
)
except SecurityError as exc:
# 安全相关错误（例如路径遍历）
raise HTTPException(
status_code=400,
detail={"message": str(exc)},
)
except ResourceNotFoundError as exc:
# 文件不存在
raise HTTPException(
status_code=404,
detail={"message": str(exc)},
)
except Exception as exc:
logger = get_server_logger()
logger.log_exception(exc, f"Unexpected error retrieving workflow args: {filename}")
# 兜底错误
raise HTTPException(
status_code=500,
detail={"message": f"Failed to retrieve workflow args: {exc}"},
)
@router.post("/api/workflows/upload/content")
async def upload_workflow_content(request: WorkflowUploadContentRequest):
return _persist_workflow_from_content(
@ -78,7 +227,7 @@ async def upload_workflow_content(request: WorkflowUploadContentRequest):
)
@router.put("/api/workflows/{filename}")
@router.put("/api/workflows/{filename}/update")
async def update_workflow_content(filename: str, request: WorkflowUpdateContentRequest):
return _persist_workflow_from_content(
filename,
@ -89,7 +238,7 @@ async def update_workflow_content(filename: str, request: WorkflowUpdateContentR
)
@router.delete("/api/workflows/{filename}")
@router.delete("/api/workflows/{filename}/delete")
async def delete_workflow(filename: str):
try:
safe_filename = validate_workflow_filename(filename, require_yaml_extension=True)
@ -180,7 +329,7 @@ async def copy_workflow_file(filename: str, request: WorkflowCopyRequest):
raise WorkflowExecutionError(f"Failed to copy workflow: {exc}")
@router.get("/api/workflows/{filename}")
@router.get("/api/workflows/{filename}/get")
async def get_workflow_raw_content(filename: str):
try:
safe_filename = validate_workflow_filename(filename, require_yaml_extension=True)
@ -209,3 +358,4 @@ async def get_workflow_raw_content(filename: str):
logger = get_server_logger()
logger.log_exception(exc, f"Unexpected error retrieving workflow: {filename}")
raise WorkflowExecutionError(f"Failed to retrieve workflow: {exc}")

25
server/services/websocket_manager.py
View File

@ -49,8 +49,6 @@ class WebSocketManager:
):
self.active_connections: Dict[str, WebSocket] = {}
self.connection_timestamps: Dict[str, float] = {}
self.send_locks: Dict[str, asyncio.Lock] = {}
self.loop: asyncio.AbstractEventLoop | None = None
self.session_store = session_store or WorkflowSessionStore()
self.session_controller = session_controller or SessionExecutionController(self.session_store)
self.attachment_service = attachment_service or AttachmentService()
@ -67,16 +65,10 @@ class WebSocketManager:
async def connect(self, websocket: WebSocket, session_id: Optional[str] = None) -> str:
await websocket.accept()
if self.loop is None:
try:
self.loop = asyncio.get_running_loop()
except RuntimeError:
self.loop = None
if not session_id:
session_id = str(uuid.uuid4())
self.active_connections[session_id] = websocket
self.connection_timestamps[session_id] = time.time()
self.send_locks[session_id] = asyncio.Lock()
logging.info("WebSocket connected: %s", session_id)
await self.send_message(
session_id,
@ -98,8 +90,6 @@ class WebSocketManager:
del self.active_connections[session_id]
if session_id in self.connection_timestamps:
del self.connection_timestamps[session_id]
if session_id in self.send_locks:
del self.send_locks[session_id]
self.session_controller.cleanup_session(session_id)
remaining_session = self.session_store.get_session(session_id)
if remaining_session and remaining_session.executor is None:
@ -111,12 +101,7 @@ class WebSocketManager:
if session_id in self.active_connections:
websocket = self.active_connections[session_id]
try:
lock = self.send_locks.get(session_id)
if lock is None:
await websocket.send_text(_encode_ws_message(message))
else:
async with lock:
await websocket.send_text(_encode_ws_message(message))
await websocket.send_text(_encode_ws_message(message))
except Exception as exc:
traceback.print_exc()
logging.error("Failed to send message to %s: %s", session_id, exc)
@ -130,13 +115,7 @@ class WebSocketManager:
else:
asyncio.run(self.send_message(session_id, message))
except RuntimeError:
if self.loop and self.loop.is_running():
asyncio.run_coroutine_threadsafe(
self.send_message(session_id, message),
self.loop,
)
else:
asyncio.run(self.send_message(session_id, message))
asyncio.run(self.send_message(session_id, message))
async def broadcast(self, message: Dict[str, Any]) -> None:
for session_id in list(self.active_connections.keys()):

21
server/services/workflow_storage.py
View File

@ -18,31 +18,16 @@ from utils.structured_logger import get_server_logger, LogType
def _update_workflow_id(content: str, workflow_id: str) -> str:
# Pattern to match graph:\n id: <value>
pattern = re.compile(r"(graph:\s*\n\s*id:\s*).*$", re.MULTILINE)
pattern = re.compile(r"^(id:\\s*).*$", re.MULTILINE)
match = pattern.search(content)
if match:
# Replace the value after "graph:\n id: "
return pattern.sub(rf"\1{workflow_id}", content, count=1)
return pattern.sub(rf"\\1{workflow_id}", content, count=1)
# If no graph.id found, look for standalone id: at root level (legacy support)
root_id_pattern = re.compile(r"^(id:\s*).*$", re.MULTILINE)
root_match = root_id_pattern.search(content)
if root_match:
return root_id_pattern.sub(rf"\1{workflow_id}", content, count=1)
# If neither found, add graph.id after graph: section if it exists
graph_pattern = re.compile(r"(graph:\s*\n)")
graph_match = graph_pattern.search(content)
if graph_match:
return graph_pattern.sub(rf"\1 id: {workflow_id}\n", content, count=1)
# Fallback (is invalid)
lines = content.splitlines()
insert_index = 0
if lines and lines[0].strip() == "---":
insert_index = 1
lines.insert(insert_index, f"graph:\n id: {workflow_id}")
lines.insert(insert_index, f"id: {workflow_id}")
updated = "\n".join(lines)
if content.endswith("\n"):
updated += "\n"

1052
uv.lock generated
View File

File diff suppressed because it is too large Load Diff