DeepSeek Harness: Python SDK: Uso Programático
Última atualização: 2026-08-31
A Web UI e o CLI são ótimos para interação humana, mas quando você precisa integrar o Agent em pipelines de automação, fluxos de CI/CD ou aplicações customizadas, o Python SDK é seu ponto de entrada — poucas linhas de código para iniciar uma sessão de Agent e obter respostas estruturadas.
📋 Pré-requisitos: Ter completado 06-tools.md, familiarizado com o sistema de ferramentas; conhecimento básico de Python
1. O Que Você Vai Aprender
- Instalação e inicialização do Python SDK
- Criação de sessões e envio de mensagens
- Obtenção de respostas do Agent e resultados de chamadas de ferramentas
- Tratamento de saída em streaming
- Interceptação e customização de chamadas de ferramentas
- Tratamento de erros e gerenciamento de timeout
2. Instalação e Inicialização
(1) Instalando o SDK
pip install deepseek-dsh
Verifique a instalação:
import dsh
print(dsh.__version__)
# 0.x.x
(2) Pré-requisitos
O Python SDK requer que o servidor DSH esteja em execução:
# Inicie o servidor DSH primeiro (modo Headless)
npx @deepseek-ai/dsh headless --port 3080
Ou especifique uma porta customizada:
npx @deepseek-ai/dsh headless --port 8080
(3) Inicializar o Cliente
▶ Exemplo 1:Criando um Cliente SDK
from dsh import DSHClient
client = DSHClient(
base_url="http://127.0.0.1:3080",
api_key="your-api-key" # Opcional, se o DSH tem autenticação configurada
)
▶ Exemplo 2:Inicializar Usando Variáveis de Ambiente
import os
from dsh import DSHClient
client = DSHClient.from_env()
# Lê as variáveis de ambiente DSH_BASE_URL e DSH_API_KEY
3. Criando uma Sessão
(1) Criando uma Nova Sessão
▶ Exemplo 3:Criando uma Sessão
session = client.create_session(
workspace="/home/alice/my-project",
model="deepseek-chat",
mode="standard"
)
print(f"Session ID: {session.id}")
print(f"Workspace: {session.workspace}")
print(f"Model: {session.model}")
(2) Configuração de Sessão
session = client.create_session(
workspace="/home/alice/my-project",
model="deepseek-chat",
mode="ptc",
sandbox="permissive",
settings={
"temperature": 0.7,
"max_tokens": 4096
}
)
(3) Restaurando uma Sessão Existente
▶ Exemplo 4:Restaurar por Session ID
session = client.get_session("sess_abc123")
print(f"Sessão restaurada: {session.id}")
print(f"Mensagens: {len(session.messages)}")
4. Enviando Mensagens e Obtendo Respostas
(1) Envio Básico de Mensagens
▶ Exemplo 5:Enviar uma Mensagem e Obter uma Resposta Completa
response = session.send("Me ajude a verificar o package.json do projeto")
print(response.content)
# O package.json do projeto mostra...
print(f"Ferramentas usadas: {len(response.tool_calls)}")
for tool in response.tool_calls:
print(f" - {tool.name}: {tool.status}")
(2) Estrutura da Resposta
class AgentResponse:
content: str # Resposta de texto do Agent
tool_calls: list[ToolCall] # Registros de chamadas de ferramentas
model: str # Modelo usado
tokens_used: int # Tokens consumidos
duration_ms: int # Tempo de resposta
class ToolCall:
name: str # Nome da ferramenta
params: dict # Parâmetros da chamada
status: str # Status de execução
result: Any # Resultado da execução
duration_ms: int # Tempo de execução
(3) Conversas Multi-turn com Contexto
▶ Exemplo 6:Conversa Multi-turn
# Primeira rodada
resp1 = session.send("Veja o conteúdo de src/app.ts")
print(resp1.content)
# Segunda rodada (o contexto é automático)
resp2 = session.send("Adicione middleware de tratamento de erros a este arquivo")
print(resp2.content)
# Terceira rodada
resp3 = session.send("Execute os testes para garantir que nada está quebrado")
print(resp3.content)
5. Chamadas de Ferramentas
(1) Chamadas Automáticas de Ferramentas
No modo Standard, o Agent decide quando chamar ferramentas automaticamente:
response = session.send("Crie src/utils/helpers.ts, escreva uma função debounce")
for tool in response.tool_calls:
print(f"Ferramenta: {tool.name}")
print(f"Parâmetros: {tool.params}")
print(f"Resultado: {tool.result}")
(2) Tratamento de Aprovação de Ferramentas
Quando uma operação do Agent requer aprovação, o SDK fornece um mecanismo de callback:
▶ Exemplo 7:Callback de Aprovação
def on_approval(tool_name: str, params: dict) -> bool:
print(f"Aprovação solicitada: {tool_name}")
print(f"Parâmetros: {params}")
# Auto-permitir operações seguras
if tool_name == "file_edit" and params.get("action") == "read":
return True
# Outras operações requerem confirmação manual
confirm = input(f"Permitir {tool_name}? (y/n): ")
return confirm.lower() == "y"
session = client.create_session(
workspace="/home/alice/project",
approval_callback=on_approval
)
(3) Desabilitando Ferramentas Específicas
session = client.create_session(
workspace="/home/alice/project",
disabled_tools=["shell", "sandbox"]
)
(4) Tratamento de Resultados de Chamadas de Ferramentas
▶ Exemplo 8:Processamento Detalhado de Resultados de Ferramentas
response = session.send("Analise a cobertura de testes do projeto")
for tool in response.tool_calls:
if tool.name == "shell":
output = tool.result.get("stdout", "")
if "Coverage" in output:
print(f"Cobertura de testes: {output}")
elif tool.name == "search":
files = tool.result.get("files", [])
print(f"Encontrados {len(files)} arquivos de teste")
elif tool.name == "file_edit":
action = tool.params.get("action")
path = tool.params.get("path")
print(f"Arquivo {action}: {path}")
6. Tratamento de Saída em Streaming
(1) Habilitando Saída em Streaming
Para respostas longas, use saída em streaming para obter resultados em tempo real:
▶ Exemplo 9:Saída em Streaming
for chunk in session.send_stream("Explique em detalhes o design de arquitetura deste projeto"):
if chunk.type == "content":
print(chunk.text, end="", flush=True)
elif chunk.type == "tool_call":
print(f"\n[Ferramenta: {chunk.tool_name}]")
elif chunk.type == "tool_result":
print(f"[Resultado da ferramenta recebido]")
(2) Tipos de Eventos de Saída em Streaming
| Tipo de Evento | Descrição | Campos de Dados |
|---|---|---|
content |
Fragmento de conteúdo de texto | text |
tool_call |
Chamada de ferramenta iniciada | tool_name, params |
tool_result |
Resultado da execução da ferramenta | tool_name, result |
approval |
Requisição de aprovação | tool_name, params |
done |
Resposta completa | tokens_used, duration_ms |
error |
Erro ocorreu | code, message |
(3) Combinando Streaming com Aprovação
▶ Exemplo 10:Tratamento de Aprovação em Saída em Streaming
def auto_approve(tool_name: str, params: dict) -> bool:
safe_actions = ["read", "search"]
if params.get("action") in safe_actions:
return True
return False
for chunk in session.send_stream(
"Refatore todos os controladores, adicione tratamento de erros",
approval_callback=auto_approve
):
if chunk.type == "content":
print(chunk.text, end="")
elif chunk.type == "approval":
print(f"\n[Auto-aprovado: {chunk.tool_name}]")
7. Recursos Avançados
(1) Integração com Modo Headless
O caso de uso mais comum do SDK é emparelhar com o modo Headless para execução autônoma do Agent:
▶ Exemplo 11:Fluxo de Trabalho Headless Completo
from dsh import DSHClient
client = DSHClient(base_url="http://127.0.0.1:3080")
def auto_approve(tool_name: str, params: dict) -> bool:
safe_tools = ["search", "file_edit", "plan"]
if tool_name in safe_tools:
action = params.get("action", "")
if action in ["read", "create"]:
return True
return False
session = client.create_session(
workspace="/home/alice/project",
model="deepseek-coder",
mode="ptc",
approval_callback=auto_approve
)
response = session.send(
"Adicione middleware de validação de entrada para todas as rotas, garantindo que os parâmetros da requisição correspondam aos tipos esperados"
)
print(f"Plano: {response.content}")
print(f"Ferramentas usadas: {len(response.tool_calls)}")
print(f"Tokens: {response.tokens_used}")
(2) Sessões Concorrentes
▶ Exemplo 12:Multi-Sessão Paralela
import concurrent.futures
def process_file(filepath: str):
client = DSHClient(base_url="http://127.0.0.1:3080")
session = client.create_session(workspace="/home/alice/project")
response = session.send(f"Adicione testes unitários para {filepath}")
return {"file": filepath, "tests_added": len(response.tool_calls)}
files = [
"src/utils/format.ts",
"src/utils/validate.ts",
"src/routes/users.ts"
]
with concurrent.futures.ThreadPoolExecutor(max_workers=3) as executor:
results = list(executor.map(process_file, files))
for r in results:
print(f"{r['file']}: {r['tests_added']} chamadas de ferramentas")
(3) Tratamento de Erros
▶ Exemplo 13:Tratamento de Erros
from dsh import DSHClient, DSHTimeoutError, DSHConnectionError
client = DSHClient(base_url="http://127.0.0.1:3080")
try:
session = client.create_session(workspace="/home/alice/project")
response = session.send("Me ajude a corrigir todos os erros TypeScript", timeout=300)
except DSHTimeoutError:
print("Timeout na resposta do Agent. Simplifique a tarefa ou aumente o timeout")
except DSHConnectionError:
print("Não é possível conectar ao servidor DSH. Verifique se está em execução")
except Exception as e:
print(f"Erro desconhecido: {e}")
8. Comparação SDK vs. Web UI/CLI
| Dimensão | Web UI | CLI | Python SDK |
|---|---|---|---|
| Método de interação | Navegador | Terminal | Código |
| Adequado para | Todos | Desenvolvedores | Engenheiros de automação |
| Mecanismo de aprovação | Interação via popup | Confirmação via linha de comando | Função callback |
| Saída em streaming | Renderização em tempo real | Saída no terminal | Fluxo de eventos |
| Concorrência | Sessão única | Sessão única | Multi-sessão |
| Capacidade de integração | Baixa | Média | Alta |
| Curva de aprendizado | Mais baixa | Baixa | Média |
❓ Perguntas Frequentes
client = DSHClient(base_url="...", debug=True). Todas as requisições e respostas HTTP serão exibidas no console.📖 Resumo
- Python SDK instala via
pip install deepseek-dsh - O SDK requer que o servidor DSH (modo Headless) esteja em execução
- Criar sessão → Enviar mensagem → Obter resposta é o fluxo central de três passos
- Aprovação de ferramentas é tratada via funções callback
- Saída em streaming
send_stream()é adequada para respostas longas - Suporta sessões concorrentes, tratamento de erros e controle de timeout
- O SDK é adequado para integração de automação; para uso diário, a Web UI é recomendada
📝 Exercícios
1. ⭐ Básico: Instale o Python SDK, inicie o DSH em modo Headless, crie uma sessão usando o SDK e envie uma mensagem "Olá", imprima o conteúdo da resposta do Agent.
2. ⭐⭐ Intermediário: Escreva um script Python que use o SDK para fazer o Agent ler o README.md do projeto e gerar um relatório de resumo do projeto, salvando-o em project-summary.txt.
3. ⭐⭐⭐ Desafio: Escreva um script de processamento em lote que use sessões concorrentes para fazer 3 Agents analisarem simultaneamente a qualidade do código em diferentes diretórios do projeto, produzindo um relatório consolidado de qualidade de código.